9bec0c329e742aaa64e883148247bd803fa27113
[fio.git] / workqueue.c
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"
10 #include "flist.h"
11 #include "workqueue.h"
12 #include "lib/getrusage.h"
13
14 enum {
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
23 static 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
42 static 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
69 static bool all_sw_idle(struct workqueue *wq)
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))
77                         return false;
78         }
79
80         return true;
81 }
82
83 /*
84  * Must be serialized wrt workqueue_enqueue() by caller
85  */
86 void 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 /*
100  * Must be serialized by caller. Returns true for queued, false for busy.
101  */
102 void workqueue_enqueue(struct workqueue *wq, struct workqueue_work *work)
103 {
104         struct submit_worker *sw;
105
106         sw = get_submit_worker(wq);
107         assert(sw);
108
109         pthread_mutex_lock(&sw->lock);
110         flist_add_tail(&work->list, &sw->work_list);
111         sw->seq = ++wq->work_seq;
112         sw->flags &= ~SW_F_IDLE;
113         pthread_mutex_unlock(&sw->lock);
114
115         pthread_cond_signal(&sw->cond);
116 }
117
118 static void handle_list(struct submit_worker *sw, struct flist_head *list)
119 {
120         struct workqueue *wq = sw->wq;
121         struct workqueue_work *work;
122
123         while (!flist_empty(list)) {
124                 work = flist_first_entry(list, struct workqueue_work, list);
125                 flist_del_init(&work->list);
126                 wq->ops.fn(sw, work);
127         }
128 }
129
130 static void *worker_thread(void *data)
131 {
132         struct submit_worker *sw = data;
133         struct workqueue *wq = sw->wq;
134         unsigned int eflags = 0, ret = 0;
135         FLIST_HEAD(local_list);
136
137         if (wq->ops.nice) {
138                 if (nice(wq->ops.nice) < 0) {
139                         log_err("workqueue: nice %s\n", strerror(errno));
140                         ret = 1;
141                 }
142         }
143
144         if (!ret)
145                 ret = workqueue_init_worker(sw);
146
147         pthread_mutex_lock(&sw->lock);
148         sw->flags |= SW_F_RUNNING;
149         if (ret)
150                 sw->flags |= SW_F_ERROR;
151         pthread_mutex_unlock(&sw->lock);
152
153         pthread_mutex_lock(&wq->flush_lock);
154         pthread_cond_signal(&wq->flush_cond);
155         pthread_mutex_unlock(&wq->flush_lock);
156
157         if (sw->flags & SW_F_ERROR)
158                 goto done;
159
160         while (1) {
161                 pthread_mutex_lock(&sw->lock);
162
163                 if (flist_empty(&sw->work_list)) {
164                         if (sw->flags & SW_F_EXIT) {
165                                 pthread_mutex_unlock(&sw->lock);
166                                 break;
167                         }
168
169                         if (workqueue_pre_sleep_check(sw)) {
170                                 pthread_mutex_unlock(&sw->lock);
171                                 workqueue_pre_sleep(sw);
172                                 pthread_mutex_lock(&sw->lock);
173                         }
174
175                         /*
176                          * We dropped and reaquired the lock, check
177                          * state again.
178                          */
179                         if (!flist_empty(&sw->work_list))
180                                 goto handle_work;
181
182                         if (sw->flags & SW_F_EXIT) {
183                                 pthread_mutex_unlock(&sw->lock);
184                                 break;
185                         } else if (!(sw->flags & SW_F_IDLE)) {
186                                 sw->flags |= SW_F_IDLE;
187                                 wq->next_free_worker = sw->index;
188                                 if (wq->wake_idle)
189                                         pthread_cond_signal(&wq->flush_cond);
190                         }
191                         if (wq->ops.update_acct_fn)
192                                 wq->ops.update_acct_fn(sw);
193
194                         pthread_cond_wait(&sw->cond, &sw->lock);
195                 } else {
196 handle_work:
197                         flist_splice_init(&sw->work_list, &local_list);
198                 }
199                 pthread_mutex_unlock(&sw->lock);
200                 handle_list(sw, &local_list);
201         }
202
203         if (wq->ops.update_acct_fn)
204                 wq->ops.update_acct_fn(sw);
205
206 done:
207         pthread_mutex_lock(&sw->lock);
208         sw->flags |= (SW_F_EXITED | eflags);
209         pthread_mutex_unlock(&sw->lock);
210         return NULL;
211 }
212
213 static void free_worker(struct submit_worker *sw, unsigned int *sum_cnt)
214 {
215         struct workqueue *wq = sw->wq;
216
217         workqueue_exit_worker(sw, sum_cnt);
218
219         pthread_cond_destroy(&sw->cond);
220         pthread_mutex_destroy(&sw->lock);
221
222         if (wq->ops.free_worker_fn)
223                 wq->ops.free_worker_fn(sw);
224 }
225
226 static void shutdown_worker(struct submit_worker *sw, unsigned int *sum_cnt)
227 {
228         pthread_join(sw->thread, NULL);
229         free_worker(sw, sum_cnt);
230 }
231
232 void workqueue_exit(struct workqueue *wq)
233 {
234         unsigned int shutdown, sum_cnt = 0;
235         struct submit_worker *sw;
236         int i;
237
238         for (i = 0; i < wq->max_workers; i++) {
239                 sw = &wq->workers[i];
240
241                 pthread_mutex_lock(&sw->lock);
242                 sw->flags |= SW_F_EXIT;
243                 pthread_cond_signal(&sw->cond);
244                 pthread_mutex_unlock(&sw->lock);
245         }
246
247         do {
248                 shutdown = 0;
249                 for (i = 0; i < wq->max_workers; i++) {
250                         sw = &wq->workers[i];
251                         if (sw->flags & SW_F_ACCOUNTED)
252                                 continue;
253                         pthread_mutex_lock(&sw->lock);
254                         sw->flags |= SW_F_ACCOUNTED;
255                         pthread_mutex_unlock(&sw->lock);
256                         shutdown_worker(sw, &sum_cnt);
257                         shutdown++;
258                 }
259         } while (shutdown && shutdown != wq->max_workers);
260
261         free(wq->workers);
262         pthread_mutex_destroy(&wq->flush_lock);
263         pthread_cond_destroy(&wq->flush_cond);
264         pthread_mutex_destroy(&wq->stat_lock);
265 }
266
267 static int start_worker(struct workqueue *wq, unsigned int index)
268 {
269         struct submit_worker *sw = &wq->workers[index];
270         int ret;
271
272         INIT_FLIST_HEAD(&sw->work_list);
273         pthread_cond_init(&sw->cond, NULL);
274         pthread_mutex_init(&sw->lock, NULL);
275         sw->wq = wq;
276         sw->index = index;
277
278         if (wq->ops.alloc_worker_fn) {
279                 ret = wq->ops.alloc_worker_fn(sw);
280                 if (ret)
281                         return ret;
282         }
283
284         ret = pthread_create(&sw->thread, NULL, worker_thread, sw);
285         if (!ret) {
286                 pthread_mutex_lock(&sw->lock);
287                 sw->flags = SW_F_IDLE;
288                 pthread_mutex_unlock(&sw->lock);
289                 return 0;
290         }
291
292         free_worker(sw, NULL);
293         return 1;
294 }
295
296 int workqueue_init(struct thread_data *td, struct workqueue *wq,
297                    struct workqueue_ops *ops, unsigned max_pending)
298 {
299         unsigned int running;
300         int i, error;
301
302         wq->max_workers = max_pending;
303         wq->td = td;
304         wq->ops = *ops;
305         wq->work_seq = 0;
306         wq->next_free_worker = 0;
307         pthread_cond_init(&wq->flush_cond, NULL);
308         pthread_mutex_init(&wq->flush_lock, NULL);
309         pthread_mutex_init(&wq->stat_lock, NULL);
310
311         wq->workers = calloc(wq->max_workers, sizeof(struct submit_worker));
312
313         for (i = 0; i < wq->max_workers; i++)
314                 if (start_worker(wq, i))
315                         break;
316
317         wq->max_workers = i;
318         if (!wq->max_workers)
319                 goto err;
320
321         /*
322          * Wait for them all to be started and initialized
323          */
324         error = 0;
325         do {
326                 struct submit_worker *sw;
327
328                 running = 0;
329                 pthread_mutex_lock(&wq->flush_lock);
330                 for (i = 0; i < wq->max_workers; i++) {
331                         sw = &wq->workers[i];
332                         pthread_mutex_lock(&sw->lock);
333                         if (sw->flags & SW_F_RUNNING)
334                                 running++;
335                         if (sw->flags & SW_F_ERROR)
336                                 error++;
337                         pthread_mutex_unlock(&sw->lock);
338                 }
339
340                 if (error || running == wq->max_workers) {
341                         pthread_mutex_unlock(&wq->flush_lock);
342                         break;
343                 }
344
345                 pthread_cond_wait(&wq->flush_cond, &wq->flush_lock);
346                 pthread_mutex_unlock(&wq->flush_lock);
347         } while (1);
348
349         if (!error)
350                 return 0;
351
352 err:
353         log_err("Can't create rate workqueue\n");
354         td_verror(td, ESRCH, "workqueue_init");
355         workqueue_exit(wq);
356         return 1;
357 }