init: use __log_buf() if we know buf != NULL
[fio.git] / workqueue.c
1 /*
2  * Generic workqueue offload mechanism
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 "smalloc.h"
13 #include "pshared.h"
14
15 enum {
16         SW_F_IDLE       = 1 << 0,
17         SW_F_RUNNING    = 1 << 1,
18         SW_F_EXIT       = 1 << 2,
19         SW_F_ACCOUNTED  = 1 << 3,
20         SW_F_ERROR      = 1 << 4,
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
114         pthread_cond_signal(&sw->cond);
115         pthread_mutex_unlock(&sw->lock);
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 ret = 0;
135         FLIST_HEAD(local_list);
136
137         sk_out_assign(sw->sk_out);
138
139         if (wq->ops.nice) {
140                 if (nice(wq->ops.nice) < 0) {
141                         log_err("workqueue: nice %s\n", strerror(errno));
142                         ret = 1;
143                 }
144         }
145
146         if (!ret)
147                 ret = workqueue_init_worker(sw);
148
149         pthread_mutex_lock(&sw->lock);
150         sw->flags |= SW_F_RUNNING;
151         if (ret)
152                 sw->flags |= SW_F_ERROR;
153         pthread_mutex_unlock(&sw->lock);
154
155         pthread_mutex_lock(&wq->flush_lock);
156         pthread_cond_signal(&wq->flush_cond);
157         pthread_mutex_unlock(&wq->flush_lock);
158
159         if (sw->flags & SW_F_ERROR)
160                 goto done;
161
162         while (1) {
163                 pthread_mutex_lock(&sw->lock);
164
165                 if (flist_empty(&sw->work_list)) {
166                         if (sw->flags & SW_F_EXIT) {
167                                 pthread_mutex_unlock(&sw->lock);
168                                 break;
169                         }
170
171                         if (workqueue_pre_sleep_check(sw)) {
172                                 pthread_mutex_unlock(&sw->lock);
173                                 workqueue_pre_sleep(sw);
174                                 pthread_mutex_lock(&sw->lock);
175                         }
176
177                         /*
178                          * We dropped and reaquired the lock, check
179                          * state again.
180                          */
181                         if (!flist_empty(&sw->work_list))
182                                 goto handle_work;
183
184                         if (sw->flags & SW_F_EXIT) {
185                                 pthread_mutex_unlock(&sw->lock);
186                                 break;
187                         } else if (!(sw->flags & SW_F_IDLE)) {
188                                 sw->flags |= SW_F_IDLE;
189                                 wq->next_free_worker = sw->index;
190                                 if (wq->wake_idle)
191                                         pthread_cond_signal(&wq->flush_cond);
192                         }
193                         if (wq->ops.update_acct_fn)
194                                 wq->ops.update_acct_fn(sw);
195
196                         pthread_cond_wait(&sw->cond, &sw->lock);
197                 } else {
198 handle_work:
199                         flist_splice_init(&sw->work_list, &local_list);
200                 }
201                 pthread_mutex_unlock(&sw->lock);
202                 handle_list(sw, &local_list);
203         }
204
205         if (wq->ops.update_acct_fn)
206                 wq->ops.update_acct_fn(sw);
207
208 done:
209         sk_out_drop();
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         if (!wq->workers)
239                 return;
240
241         for (i = 0; i < wq->max_workers; i++) {
242                 sw = &wq->workers[i];
243
244                 pthread_mutex_lock(&sw->lock);
245                 sw->flags |= SW_F_EXIT;
246                 pthread_cond_signal(&sw->cond);
247                 pthread_mutex_unlock(&sw->lock);
248         }
249
250         do {
251                 shutdown = 0;
252                 for (i = 0; i < wq->max_workers; i++) {
253                         sw = &wq->workers[i];
254                         if (sw->flags & SW_F_ACCOUNTED)
255                                 continue;
256                         pthread_mutex_lock(&sw->lock);
257                         sw->flags |= SW_F_ACCOUNTED;
258                         pthread_mutex_unlock(&sw->lock);
259                         shutdown_worker(sw, &sum_cnt);
260                         shutdown++;
261                 }
262         } while (shutdown && shutdown != wq->max_workers);
263
264         sfree(wq->workers);
265         wq->workers = NULL;
266         pthread_mutex_destroy(&wq->flush_lock);
267         pthread_cond_destroy(&wq->flush_cond);
268         pthread_mutex_destroy(&wq->stat_lock);
269 }
270
271 static int start_worker(struct workqueue *wq, unsigned int index,
272                         struct sk_out *sk_out)
273 {
274         struct submit_worker *sw = &wq->workers[index];
275         int ret;
276
277         INIT_FLIST_HEAD(&sw->work_list);
278
279         ret = mutex_cond_init_pshared(&sw->lock, &sw->cond);
280         if (ret)
281                 return ret;
282
283         sw->wq = wq;
284         sw->index = index;
285         sw->sk_out = sk_out;
286
287         if (wq->ops.alloc_worker_fn) {
288                 ret = wq->ops.alloc_worker_fn(sw);
289                 if (ret)
290                         return ret;
291         }
292
293         ret = pthread_create(&sw->thread, NULL, worker_thread, sw);
294         if (!ret) {
295                 pthread_mutex_lock(&sw->lock);
296                 sw->flags = SW_F_IDLE;
297                 pthread_mutex_unlock(&sw->lock);
298                 return 0;
299         }
300
301         free_worker(sw, NULL);
302         return 1;
303 }
304
305 int workqueue_init(struct thread_data *td, struct workqueue *wq,
306                    struct workqueue_ops *ops, unsigned int max_workers,
307                    struct sk_out *sk_out)
308 {
309         unsigned int running;
310         int i, error;
311         int ret;
312
313         wq->max_workers = max_workers;
314         wq->td = td;
315         wq->ops = *ops;
316         wq->work_seq = 0;
317         wq->next_free_worker = 0;
318
319         ret = mutex_cond_init_pshared(&wq->flush_lock, &wq->flush_cond);
320         if (ret)
321                 goto err;
322         ret = mutex_init_pshared(&wq->stat_lock);
323         if (ret)
324                 goto err;
325
326         wq->workers = smalloc(wq->max_workers * sizeof(struct submit_worker));
327         if (!wq->workers)
328                 goto err;
329
330         for (i = 0; i < wq->max_workers; i++)
331                 if (start_worker(wq, i, sk_out))
332                         break;
333
334         wq->max_workers = i;
335         if (!wq->max_workers)
336                 goto err;
337
338         /*
339          * Wait for them all to be started and initialized
340          */
341         error = 0;
342         do {
343                 struct submit_worker *sw;
344
345                 running = 0;
346                 pthread_mutex_lock(&wq->flush_lock);
347                 for (i = 0; i < wq->max_workers; i++) {
348                         sw = &wq->workers[i];
349                         pthread_mutex_lock(&sw->lock);
350                         if (sw->flags & SW_F_RUNNING)
351                                 running++;
352                         if (sw->flags & SW_F_ERROR)
353                                 error++;
354                         pthread_mutex_unlock(&sw->lock);
355                 }
356
357                 if (error || running == wq->max_workers) {
358                         pthread_mutex_unlock(&wq->flush_lock);
359                         break;
360                 }
361
362                 pthread_cond_wait(&wq->flush_cond, &wq->flush_lock);
363                 pthread_mutex_unlock(&wq->flush_lock);
364         } while (1);
365
366         if (!error)
367                 return 0;
368
369 err:
370         log_err("Can't create rate workqueue\n");
371         td_verror(td, ESRCH, "workqueue_init");
372         workqueue_exit(wq);
373         return 1;
374 }