workqueue: move 'td' private data to the workqueue user
[fio.git] / workqueue.c
CommitLineData
a9da8ab2
JA
1/*
2 * Rated submission helpers
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"
12#include "lib/getrusage.h"
13
a9da8ab2
JA
14enum {
15 SW_F_IDLE = 1 << 0,
16 SW_F_RUNNING = 1 << 1,
17 SW_F_EXIT = 1 << 2,
18 SW_F_EXITED = 1 << 3,
19 SW_F_ACCOUNTED = 1 << 4,
20 SW_F_ERROR = 1 << 5,
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{
88 wq->wake_idle = 1;
89
90 while (!all_sw_idle(wq)) {
91 pthread_mutex_lock(&wq->flush_lock);
92 pthread_cond_wait(&wq->flush_cond, &wq->flush_lock);
93 pthread_mutex_unlock(&wq->flush_lock);
94 }
95
96 wq->wake_idle = 0;
97}
98
99/*
b07f6ad1 100 * Must be serialized by caller. Returns true for queued, false for busy.
a9da8ab2 101 */
88271841 102bool workqueue_enqueue(struct workqueue *wq, struct workqueue_work *work)
a9da8ab2
JA
103{
104 struct submit_worker *sw;
105
106 sw = get_submit_worker(wq);
107 if (sw) {
a9da8ab2 108 pthread_mutex_lock(&sw->lock);
88271841 109 flist_add_tail(&work->list, &sw->work_list);
a9da8ab2
JA
110 sw->seq = ++wq->work_seq;
111 sw->flags &= ~SW_F_IDLE;
112 pthread_mutex_unlock(&sw->lock);
113
114 pthread_cond_signal(&sw->cond);
b07f6ad1 115 return true;
a9da8ab2
JA
116 }
117
b07f6ad1 118 return false;
a9da8ab2
JA
119}
120
121static void handle_list(struct submit_worker *sw, struct flist_head *list)
122{
123 struct workqueue *wq = sw->wq;
88271841 124 struct workqueue_work *work;
a9da8ab2
JA
125
126 while (!flist_empty(list)) {
88271841
JA
127 work = flist_first_entry(list, struct workqueue_work, list);
128 flist_del_init(&work->list);
ee2b6d6e 129 wq->ops.fn(sw, work);
a9da8ab2
JA
130 }
131}
132
133static int init_submit_worker(struct submit_worker *sw)
134{
135 struct thread_data *parent = sw->wq->td;
ee2b6d6e 136 struct thread_data *td = sw->private;
a9da8ab2
JA
137 int fio_unused ret;
138
139 memcpy(&td->o, &parent->o, sizeof(td->o));
140 memcpy(&td->ts, &parent->ts, sizeof(td->ts));
141 td->o.uid = td->o.gid = -1U;
142 dup_files(td, parent);
736a50cd 143 td->eo = parent->eo;
a9da8ab2
JA
144 fio_options_mem_dupe(td);
145
146 if (ioengine_load(td))
147 goto err;
148
149 if (td->o.odirect)
150 td->io_ops->flags |= FIO_RAWIO;
151
152 td->pid = gettid();
153
154 INIT_FLIST_HEAD(&td->io_log_list);
155 INIT_FLIST_HEAD(&td->io_hist_list);
156 INIT_FLIST_HEAD(&td->verify_list);
157 INIT_FLIST_HEAD(&td->trim_list);
158 INIT_FLIST_HEAD(&td->next_rand_list);
159 td->io_hist_tree = RB_ROOT;
160
161 td->o.iodepth = 1;
162 if (td_io_init(td))
163 goto err_io_init;
164
165 fio_gettime(&td->epoch, NULL);
166 fio_getrusage(&td->ru_start);
ac28d905 167 clear_io_state(td, 1);
a9da8ab2
JA
168
169 td_set_runstate(td, TD_RUNNING);
170 td->flags |= TD_F_CHILD;
171 td->parent = parent;
172 return 0;
173
174err_io_init:
175 close_ioengine(td);
176err:
177 return 1;
178}
179
f5cd2907 180#ifdef CONFIG_SFAA
a9da8ab2
JA
181static void sum_val(uint64_t *dst, uint64_t *src)
182{
183 if (*src) {
184 __sync_fetch_and_add(dst, *src);
185 *src = 0;
186 }
187}
2a274336
JA
188#else
189static void sum_val(uint64_t *dst, uint64_t *src)
190{
191 if (*src) {
192 *dst += *src;
193 *src = 0;
194 }
195}
196#endif
a9da8ab2 197
a6a3469e
JA
198static void pthread_double_unlock(pthread_mutex_t *lock1,
199 pthread_mutex_t *lock2)
a9da8ab2 200{
f5cd2907 201#ifndef CONFIG_SFAA
a6a3469e
JA
202 pthread_mutex_unlock(lock1);
203 pthread_mutex_unlock(lock2);
2a274336 204#endif
a6a3469e
JA
205}
206
207static void pthread_double_lock(pthread_mutex_t *lock1, pthread_mutex_t *lock2)
208{
209#ifndef CONFIG_SFAA
210 if (lock1 < lock2) {
211 pthread_mutex_lock(lock1);
212 pthread_mutex_lock(lock2);
213 } else {
214 pthread_mutex_lock(lock2);
215 pthread_mutex_lock(lock1);
216 }
217#endif
218}
219
220static void sum_ddir(struct thread_data *dst, struct thread_data *src,
221 enum fio_ddir ddir)
222{
223 pthread_double_lock(&dst->io_wq.stat_lock, &src->io_wq.stat_lock);
2a274336 224
a9da8ab2
JA
225 sum_val(&dst->io_bytes[ddir], &src->io_bytes[ddir]);
226 sum_val(&dst->io_blocks[ddir], &src->io_blocks[ddir]);
227 sum_val(&dst->this_io_blocks[ddir], &src->this_io_blocks[ddir]);
228 sum_val(&dst->this_io_bytes[ddir], &src->this_io_bytes[ddir]);
229 sum_val(&dst->bytes_done[ddir], &src->bytes_done[ddir]);
2a274336 230
a6a3469e 231 pthread_double_unlock(&dst->io_wq.stat_lock, &src->io_wq.stat_lock);
a9da8ab2
JA
232}
233
234static void update_accounting(struct submit_worker *sw)
235{
ee2b6d6e 236 struct thread_data *src = sw->private;
a9da8ab2
JA
237 struct thread_data *dst = sw->wq->td;
238
239 if (td_read(src))
240 sum_ddir(dst, src, DDIR_READ);
241 if (td_write(src))
242 sum_ddir(dst, src, DDIR_WRITE);
243 if (td_trim(src))
244 sum_ddir(dst, src, DDIR_TRIM);
245}
246
247static void *worker_thread(void *data)
248{
249 struct submit_worker *sw = data;
250 struct workqueue *wq = sw->wq;
a9da8ab2
JA
251 unsigned int eflags = 0, ret;
252 FLIST_HEAD(local_list);
253
254 ret = init_submit_worker(sw);
255 pthread_mutex_lock(&sw->lock);
256 sw->flags |= SW_F_RUNNING;
257 if (ret)
258 sw->flags |= SW_F_ERROR;
259 pthread_mutex_unlock(&sw->lock);
260
261 pthread_mutex_lock(&wq->flush_lock);
262 pthread_cond_signal(&wq->flush_cond);
263 pthread_mutex_unlock(&wq->flush_lock);
264
265 if (sw->flags & SW_F_ERROR)
266 goto done;
267
268 while (1) {
269 pthread_mutex_lock(&sw->lock);
270
271 if (flist_empty(&sw->work_list)) {
272 if (sw->flags & SW_F_EXIT) {
273 pthread_mutex_unlock(&sw->lock);
274 break;
275 }
276
ee2b6d6e 277 if (workqueue_pre_sleep_check(sw)) {
a9da8ab2 278 pthread_mutex_unlock(&sw->lock);
ee2b6d6e 279 workqueue_pre_sleep(sw);
a9da8ab2
JA
280 pthread_mutex_lock(&sw->lock);
281 }
282
283 /*
284 * We dropped and reaquired the lock, check
285 * state again.
286 */
287 if (!flist_empty(&sw->work_list))
288 goto handle_work;
289
290 if (sw->flags & SW_F_EXIT) {
291 pthread_mutex_unlock(&sw->lock);
292 break;
293 } else if (!(sw->flags & SW_F_IDLE)) {
294 sw->flags |= SW_F_IDLE;
295 wq->next_free_worker = sw->index;
296 if (wq->wake_idle)
297 pthread_cond_signal(&wq->flush_cond);
298 }
299 update_accounting(sw);
300 pthread_cond_wait(&sw->cond, &sw->lock);
301 } else {
302handle_work:
303 flist_splice_init(&sw->work_list, &local_list);
304 }
305 pthread_mutex_unlock(&sw->lock);
306 handle_list(sw, &local_list);
307 }
308
309 update_accounting(sw);
310
311done:
312 pthread_mutex_lock(&sw->lock);
313 sw->flags |= (SW_F_EXITED | eflags);
314 pthread_mutex_unlock(&sw->lock);
315 return NULL;
316}
317
318static void free_worker(struct submit_worker *sw)
319{
ee2b6d6e
JA
320 struct thread_data *td = sw->private;
321 struct workqueue *wq = sw->wq;
a9da8ab2
JA
322
323 fio_options_free(td);
324 close_and_free_files(td);
325 if (td->io_ops)
326 close_ioengine(td);
327 td_set_runstate(td, TD_EXITED);
328
329 pthread_cond_destroy(&sw->cond);
330 pthread_mutex_destroy(&sw->lock);
ee2b6d6e
JA
331
332 if (wq->ops.free_worker_fn)
333 wq->ops.free_worker_fn(sw);
a9da8ab2
JA
334}
335
336static void shutdown_worker(struct submit_worker *sw, unsigned int *sum_cnt)
337{
338 struct thread_data *parent = sw->wq->td;
ee2b6d6e 339 struct thread_data *td = sw->private;
a9da8ab2
JA
340
341 pthread_join(sw->thread, NULL);
342 (*sum_cnt)++;
ee2b6d6e 343 sum_thread_stats(&parent->ts, &td->ts, *sum_cnt == 1);
a9da8ab2
JA
344 free_worker(sw);
345}
346
347void workqueue_exit(struct workqueue *wq)
348{
349 unsigned int shutdown, sum_cnt = 0;
350 struct submit_worker *sw;
351 int i;
352
353 for (i = 0; i < wq->max_workers; i++) {
354 sw = &wq->workers[i];
355
356 pthread_mutex_lock(&sw->lock);
357 sw->flags |= SW_F_EXIT;
358 pthread_cond_signal(&sw->cond);
359 pthread_mutex_unlock(&sw->lock);
360 }
361
362 do {
363 shutdown = 0;
364 for (i = 0; i < wq->max_workers; i++) {
365 sw = &wq->workers[i];
366 if (sw->flags & SW_F_ACCOUNTED)
367 continue;
b7d0bbf1 368 pthread_mutex_lock(&sw->lock);
a9da8ab2 369 sw->flags |= SW_F_ACCOUNTED;
b7d0bbf1 370 pthread_mutex_unlock(&sw->lock);
a9da8ab2
JA
371 shutdown_worker(sw, &sum_cnt);
372 shutdown++;
373 }
374 } while (shutdown && shutdown != wq->max_workers);
375
376 free(wq->workers);
377 pthread_mutex_destroy(&wq->flush_lock);
378 pthread_cond_destroy(&wq->flush_cond);
2a274336 379 pthread_mutex_destroy(&wq->stat_lock);
a9da8ab2
JA
380}
381
382static int start_worker(struct workqueue *wq, unsigned int index)
383{
384 struct submit_worker *sw = &wq->workers[index];
385 int ret;
386
387 INIT_FLIST_HEAD(&sw->work_list);
388 pthread_cond_init(&sw->cond, NULL);
389 pthread_mutex_init(&sw->lock, NULL);
390 sw->wq = wq;
391 sw->index = index;
392
ee2b6d6e
JA
393 if (wq->ops.alloc_worker_fn) {
394 ret = wq->ops.alloc_worker_fn(sw);
395 if (ret)
396 return ret;
397 }
398
a9da8ab2
JA
399 ret = pthread_create(&sw->thread, NULL, worker_thread, sw);
400 if (!ret) {
401 pthread_mutex_lock(&sw->lock);
402 sw->flags = SW_F_IDLE;
403 pthread_mutex_unlock(&sw->lock);
404 return 0;
405 }
406
407 free_worker(sw);
408 return 1;
409}
410
411int workqueue_init(struct thread_data *td, struct workqueue *wq,
5bb79f69 412 struct workqueue_ops *ops, unsigned max_pending)
a9da8ab2
JA
413{
414 unsigned int running;
415 int i, error;
416
417 wq->max_workers = max_pending;
418 wq->td = td;
5bb79f69 419 wq->ops = *ops;
a9da8ab2
JA
420 wq->work_seq = 0;
421 wq->next_free_worker = 0;
422 pthread_cond_init(&wq->flush_cond, NULL);
423 pthread_mutex_init(&wq->flush_lock, NULL);
2a274336 424 pthread_mutex_init(&wq->stat_lock, NULL);
a9da8ab2
JA
425
426 wq->workers = calloc(wq->max_workers, sizeof(struct submit_worker));
427
428 for (i = 0; i < wq->max_workers; i++)
429 if (start_worker(wq, i))
430 break;
431
432 wq->max_workers = i;
bddc8d16
JA
433 if (!wq->max_workers)
434 goto err;
a9da8ab2
JA
435
436 /*
437 * Wait for them all to be started and initialized
438 */
439 error = 0;
440 do {
441 struct submit_worker *sw;
442
443 running = 0;
444 pthread_mutex_lock(&wq->flush_lock);
445 for (i = 0; i < wq->max_workers; i++) {
446 sw = &wq->workers[i];
447 pthread_mutex_lock(&sw->lock);
448 if (sw->flags & SW_F_RUNNING)
449 running++;
450 if (sw->flags & SW_F_ERROR)
451 error++;
452 pthread_mutex_unlock(&sw->lock);
453 }
454
455 if (error || running == wq->max_workers) {
456 pthread_mutex_unlock(&wq->flush_lock);
457 break;
458 }
459
460 pthread_cond_wait(&wq->flush_cond, &wq->flush_lock);
461 pthread_mutex_unlock(&wq->flush_lock);
462 } while (1);
463
bddc8d16
JA
464 if (!error)
465 return 0;
a9da8ab2 466
bddc8d16
JA
467err:
468 log_err("Can't create rate workqueue\n");
469 td_verror(td, ESRCH, "workqueue_init");
470 workqueue_exit(wq);
471 return 1;
a9da8ab2 472}