workqueue: Rework while loop locking
[fio.git] / workqueue.c
CommitLineData
a9da8ab2 1/*
51575029 2 * Generic workqueue offload mechanism
a9da8ab2
JA
3 *
4 * Copyright (C) 2015 Jens Axboe <axboe@kernel.dk>
5 *
6 */
7#include <unistd.h>
8
9#include "fio.h"
a9da8ab2
JA
10#include "flist.h"
11#include "workqueue.h"
a47591e4 12#include "smalloc.h"
ae626d4e 13#include "pshared.h"
a9da8ab2 14
a9da8ab2
JA
15enum {
16 SW_F_IDLE = 1 << 0,
17 SW_F_RUNNING = 1 << 1,
18 SW_F_EXIT = 1 << 2,
d22042d2
JA
19 SW_F_ACCOUNTED = 1 << 3,
20 SW_F_ERROR = 1 << 4,
a9da8ab2
JA
21};
22
23static struct submit_worker *__get_submit_worker(struct workqueue *wq,
24 unsigned int start,
25 unsigned int end,
26 struct submit_worker **best)
27{
28 struct submit_worker *sw = NULL;
29
30 while (start <= end) {
31 sw = &wq->workers[start];
32 if (sw->flags & SW_F_IDLE)
33 return sw;
34 if (!(*best) || sw->seq < (*best)->seq)
35 *best = sw;
36 start++;
37 }
38
39 return NULL;
40}
41
42static struct submit_worker *get_submit_worker(struct workqueue *wq)
43{
44 unsigned int next = wq->next_free_worker;
45 struct submit_worker *sw, *best = NULL;
46
47 assert(next < wq->max_workers);
48
49 sw = __get_submit_worker(wq, next, wq->max_workers - 1, &best);
50 if (!sw && next)
51 sw = __get_submit_worker(wq, 0, next - 1, &best);
52
53 /*
54 * No truly idle found, use best match
55 */
56 if (!sw)
57 sw = best;
58
59 if (sw->index == wq->next_free_worker) {
60 if (sw->index + 1 < wq->max_workers)
61 wq->next_free_worker = sw->index + 1;
62 else
63 wq->next_free_worker = 0;
64 }
65
66 return sw;
67}
68
1391052a 69static bool all_sw_idle(struct workqueue *wq)
a9da8ab2
JA
70{
71 int i;
72
73 for (i = 0; i < wq->max_workers; i++) {
74 struct submit_worker *sw = &wq->workers[i];
75
76 if (!(sw->flags & SW_F_IDLE))
1391052a 77 return false;
a9da8ab2
JA
78 }
79
1391052a 80 return true;
a9da8ab2
JA
81}
82
83/*
84 * Must be serialized wrt workqueue_enqueue() by caller
85 */
86void workqueue_flush(struct workqueue *wq)
87{
2dddd728 88 pthread_mutex_lock(&wq->flush_lock);
a9da8ab2
JA
89 wq->wake_idle = 1;
90
2dddd728 91 while (!all_sw_idle(wq))
a9da8ab2 92 pthread_cond_wait(&wq->flush_cond, &wq->flush_lock);
a9da8ab2
JA
93
94 wq->wake_idle = 0;
2dddd728 95 pthread_mutex_unlock(&wq->flush_lock);
a9da8ab2
JA
96}
97
98/*
0943e13c 99 * Must be serialized by caller.
a9da8ab2 100 */
26de50cf 101void workqueue_enqueue(struct workqueue *wq, struct workqueue_work *work)
a9da8ab2
JA
102{
103 struct submit_worker *sw;
104
105 sw = get_submit_worker(wq);
26de50cf
JA
106 assert(sw);
107
108 pthread_mutex_lock(&sw->lock);
109 flist_add_tail(&work->list, &sw->work_list);
110 sw->seq = ++wq->work_seq;
111 sw->flags &= ~SW_F_IDLE;
a9da8ab2 112
26de50cf 113 pthread_cond_signal(&sw->cond);
b93a8eb0 114 pthread_mutex_unlock(&sw->lock);
a9da8ab2
JA
115}
116
117static void handle_list(struct submit_worker *sw, struct flist_head *list)
118{
119 struct workqueue *wq = sw->wq;
88271841 120 struct workqueue_work *work;
a9da8ab2
JA
121
122 while (!flist_empty(list)) {
88271841
JA
123 work = flist_first_entry(list, struct workqueue_work, list);
124 flist_del_init(&work->list);
ee2b6d6e 125 wq->ops.fn(sw, work);
a9da8ab2
JA
126 }
127}
128
a9da8ab2
JA
129static void *worker_thread(void *data)
130{
131 struct submit_worker *sw = data;
132 struct workqueue *wq = sw->wq;
d22042d2 133 unsigned int ret = 0;
a9da8ab2
JA
134 FLIST_HEAD(local_list);
135
24660963
JA
136 sk_out_assign(sw->sk_out);
137
f6496ba7
JA
138 if (wq->ops.nice) {
139 if (nice(wq->ops.nice) < 0) {
140 log_err("workqueue: nice %s\n", strerror(errno));
141 ret = 1;
142 }
143 }
144
145 if (!ret)
146 ret = workqueue_init_worker(sw);
147
a9da8ab2
JA
148 pthread_mutex_lock(&sw->lock);
149 sw->flags |= SW_F_RUNNING;
150 if (ret)
151 sw->flags |= SW_F_ERROR;
152 pthread_mutex_unlock(&sw->lock);
153
154 pthread_mutex_lock(&wq->flush_lock);
155 pthread_cond_signal(&wq->flush_cond);
156 pthread_mutex_unlock(&wq->flush_lock);
157
158 if (sw->flags & SW_F_ERROR)
159 goto done;
160
2dddd728 161 pthread_mutex_lock(&sw->lock);
a9da8ab2 162 while (1) {
a9da8ab2
JA
163 if (flist_empty(&sw->work_list)) {
164 if (sw->flags & SW_F_EXIT) {
a9da8ab2
JA
165 break;
166 }
167
ee2b6d6e 168 if (workqueue_pre_sleep_check(sw)) {
a9da8ab2 169 pthread_mutex_unlock(&sw->lock);
ee2b6d6e 170 workqueue_pre_sleep(sw);
a9da8ab2
JA
171 pthread_mutex_lock(&sw->lock);
172 }
173
174 /*
175 * We dropped and reaquired the lock, check
176 * state again.
177 */
178 if (!flist_empty(&sw->work_list))
179 goto handle_work;
180
181 if (sw->flags & SW_F_EXIT) {
a9da8ab2
JA
182 break;
183 } else if (!(sw->flags & SW_F_IDLE)) {
184 sw->flags |= SW_F_IDLE;
185 wq->next_free_worker = sw->index;
186 if (wq->wake_idle)
187 pthread_cond_signal(&wq->flush_cond);
188 }
17ecadc3 189
a9da8ab2
JA
190 pthread_cond_wait(&sw->cond, &sw->lock);
191 } else {
192handle_work:
193 flist_splice_init(&sw->work_list, &local_list);
194 }
195 pthread_mutex_unlock(&sw->lock);
196 handle_list(sw, &local_list);
4cf30b66
VF
197 if (wq->ops.update_acct_fn)
198 wq->ops.update_acct_fn(sw);
2dddd728 199 pthread_mutex_lock(&sw->lock);
a9da8ab2 200 }
2dddd728 201 pthread_mutex_unlock(&sw->lock);
a9da8ab2 202
a9da8ab2 203done:
24660963 204 sk_out_drop();
a9da8ab2
JA
205 return NULL;
206}
207
c35c582d 208static void free_worker(struct submit_worker *sw, unsigned int *sum_cnt)
a9da8ab2 209{
ee2b6d6e 210 struct workqueue *wq = sw->wq;
a9da8ab2 211
c35c582d 212 workqueue_exit_worker(sw, sum_cnt);
a9da8ab2
JA
213
214 pthread_cond_destroy(&sw->cond);
215 pthread_mutex_destroy(&sw->lock);
ee2b6d6e
JA
216
217 if (wq->ops.free_worker_fn)
218 wq->ops.free_worker_fn(sw);
a9da8ab2
JA
219}
220
221static void shutdown_worker(struct submit_worker *sw, unsigned int *sum_cnt)
222{
a9da8ab2 223 pthread_join(sw->thread, NULL);
c35c582d 224 free_worker(sw, sum_cnt);
a9da8ab2
JA
225}
226
227void workqueue_exit(struct workqueue *wq)
228{
229 unsigned int shutdown, sum_cnt = 0;
230 struct submit_worker *sw;
231 int i;
232
f8e266c9
JA
233 if (!wq->workers)
234 return;
235
a9da8ab2
JA
236 for (i = 0; i < wq->max_workers; i++) {
237 sw = &wq->workers[i];
238
239 pthread_mutex_lock(&sw->lock);
240 sw->flags |= SW_F_EXIT;
241 pthread_cond_signal(&sw->cond);
242 pthread_mutex_unlock(&sw->lock);
243 }
244
245 do {
246 shutdown = 0;
247 for (i = 0; i < wq->max_workers; i++) {
248 sw = &wq->workers[i];
249 if (sw->flags & SW_F_ACCOUNTED)
250 continue;
b7d0bbf1 251 pthread_mutex_lock(&sw->lock);
a9da8ab2 252 sw->flags |= SW_F_ACCOUNTED;
b7d0bbf1 253 pthread_mutex_unlock(&sw->lock);
a9da8ab2
JA
254 shutdown_worker(sw, &sum_cnt);
255 shutdown++;
256 }
257 } while (shutdown && shutdown != wq->max_workers);
258
a47591e4 259 sfree(wq->workers);
f8e266c9 260 wq->workers = NULL;
a9da8ab2
JA
261 pthread_mutex_destroy(&wq->flush_lock);
262 pthread_cond_destroy(&wq->flush_cond);
2a274336 263 pthread_mutex_destroy(&wq->stat_lock);
a9da8ab2
JA
264}
265
24660963
JA
266static int start_worker(struct workqueue *wq, unsigned int index,
267 struct sk_out *sk_out)
a9da8ab2
JA
268{
269 struct submit_worker *sw = &wq->workers[index];
270 int ret;
271
272 INIT_FLIST_HEAD(&sw->work_list);
34febb23
JA
273
274 ret = mutex_cond_init_pshared(&sw->lock, &sw->cond);
f9e5b5ee
JK
275 if (ret)
276 return ret;
34febb23 277
a9da8ab2
JA
278 sw->wq = wq;
279 sw->index = index;
24660963 280 sw->sk_out = sk_out;
a9da8ab2 281
ee2b6d6e
JA
282 if (wq->ops.alloc_worker_fn) {
283 ret = wq->ops.alloc_worker_fn(sw);
284 if (ret)
285 return ret;
286 }
287
a9da8ab2
JA
288 ret = pthread_create(&sw->thread, NULL, worker_thread, sw);
289 if (!ret) {
290 pthread_mutex_lock(&sw->lock);
291 sw->flags = SW_F_IDLE;
292 pthread_mutex_unlock(&sw->lock);
293 return 0;
294 }
295
c35c582d 296 free_worker(sw, NULL);
a9da8ab2
JA
297 return 1;
298}
299
300int workqueue_init(struct thread_data *td, struct workqueue *wq,
24660963
JA
301 struct workqueue_ops *ops, unsigned int max_workers,
302 struct sk_out *sk_out)
a9da8ab2
JA
303{
304 unsigned int running;
305 int i, error;
f9e5b5ee 306 int ret;
a9da8ab2 307
24660963 308 wq->max_workers = max_workers;
a9da8ab2 309 wq->td = td;
5bb79f69 310 wq->ops = *ops;
a9da8ab2
JA
311 wq->work_seq = 0;
312 wq->next_free_worker = 0;
f9e5b5ee 313
34febb23
JA
314 ret = mutex_cond_init_pshared(&wq->flush_lock, &wq->flush_cond);
315 if (ret)
f9e5b5ee 316 goto err;
34febb23
JA
317 ret = mutex_init_pshared(&wq->stat_lock);
318 if (ret)
f9e5b5ee 319 goto err;
a9da8ab2 320
a47591e4 321 wq->workers = smalloc(wq->max_workers * sizeof(struct submit_worker));
81b3c86f
JA
322 if (!wq->workers)
323 goto err;
a9da8ab2
JA
324
325 for (i = 0; i < wq->max_workers; i++)
24660963 326 if (start_worker(wq, i, sk_out))
a9da8ab2
JA
327 break;
328
329 wq->max_workers = i;
bddc8d16
JA
330 if (!wq->max_workers)
331 goto err;
a9da8ab2
JA
332
333 /*
334 * Wait for them all to be started and initialized
335 */
336 error = 0;
2dddd728 337 pthread_mutex_lock(&wq->flush_lock);
a9da8ab2
JA
338 do {
339 struct submit_worker *sw;
340
341 running = 0;
a9da8ab2
JA
342 for (i = 0; i < wq->max_workers; i++) {
343 sw = &wq->workers[i];
344 pthread_mutex_lock(&sw->lock);
345 if (sw->flags & SW_F_RUNNING)
346 running++;
347 if (sw->flags & SW_F_ERROR)
348 error++;
349 pthread_mutex_unlock(&sw->lock);
350 }
351
2dddd728 352 if (error || running == wq->max_workers)
a9da8ab2 353 break;
a9da8ab2
JA
354
355 pthread_cond_wait(&wq->flush_cond, &wq->flush_lock);
a9da8ab2 356 } while (1);
2dddd728 357 pthread_mutex_unlock(&wq->flush_lock);
a9da8ab2 358
bddc8d16
JA
359 if (!error)
360 return 0;
a9da8ab2 361
bddc8d16
JA
362err:
363 log_err("Can't create rate workqueue\n");
364 td_verror(td, ESRCH, "workqueue_init");
365 workqueue_exit(wq);
366 return 1;
a9da8ab2 367}