3 #include "lib/getrusage.h"
5 static void io_workqueue_fn(struct submit_worker *sw,
6 struct workqueue_work *work)
8 struct io_u *io_u = container_of(work, struct io_u, work);
9 const enum fio_ddir ddir = io_u->ddir;
10 struct thread_data *td = sw->private;
13 dprint(FD_RATE, "io_u %p queued by %u\n", io_u, gettid());
15 io_u_set(io_u, IO_U_F_NO_FILE_PUT);
20 ret = td_io_queue(td, io_u);
21 if (ret != FIO_Q_BUSY)
23 ret = io_u_queued_complete(td, 1);
26 io_u_clear(io_u, IO_U_F_FLIGHT);
29 dprint(FD_RATE, "io_u %p ret %d by %u\n", io_u, ret, gettid());
31 io_queue_event(td, io_u, &ret, ddir, NULL, 0, NULL);
33 if (ret == FIO_Q_COMPLETED)
35 else if (ret == FIO_Q_QUEUED) {
36 unsigned int min_evts;
38 if (td->o.iodepth == 1)
43 ret = io_u_queued_complete(td, min_evts);
46 } else if (ret == FIO_Q_BUSY) {
47 ret = io_u_queued_complete(td, td->cur_depth);
53 static bool io_workqueue_pre_sleep_flush_fn(struct submit_worker *sw)
55 struct thread_data *td = sw->private;
57 if (td->io_u_queued || td->cur_depth || td->io_u_in_flight)
63 static void io_workqueue_pre_sleep_fn(struct submit_worker *sw)
65 struct thread_data *td = sw->private;
68 ret = io_u_quiesce(td);
73 static int io_workqueue_alloc_fn(struct submit_worker *sw)
75 struct thread_data *td;
77 td = calloc(1, sizeof(*td));
82 static void io_workqueue_free_fn(struct submit_worker *sw)
88 static int io_workqueue_init_worker_fn(struct submit_worker *sw)
90 struct thread_data *parent = sw->wq->td;
91 struct thread_data *td = sw->private;
94 memcpy(&td->o, &parent->o, sizeof(td->o));
95 memcpy(&td->ts, &parent->ts, sizeof(td->ts));
96 td->o.uid = td->o.gid = -1U;
97 dup_files(td, parent);
99 fio_options_mem_dupe(td);
101 if (ioengine_load(td))
105 td->io_ops->flags |= FIO_RAWIO;
109 INIT_FLIST_HEAD(&td->io_log_list);
110 INIT_FLIST_HEAD(&td->io_hist_list);
111 INIT_FLIST_HEAD(&td->verify_list);
112 INIT_FLIST_HEAD(&td->trim_list);
113 INIT_FLIST_HEAD(&td->next_rand_list);
114 td->io_hist_tree = RB_ROOT;
120 fio_gettime(&td->epoch, NULL);
121 fio_getrusage(&td->ru_start);
122 clear_io_state(td, 1);
124 td_set_runstate(td, TD_RUNNING);
125 td->flags |= TD_F_CHILD;
136 static void io_workqueue_exit_worker_fn(struct submit_worker *sw,
137 unsigned int *sum_cnt)
139 struct thread_data *td = sw->private;
142 sum_thread_stats(&sw->wq->td->ts, &td->ts, *sum_cnt == 1);
144 fio_options_free(td);
145 close_and_free_files(td);
148 td_set_runstate(td, TD_EXITED);
152 static void sum_val(uint64_t *dst, uint64_t *src)
155 __sync_fetch_and_add(dst, *src);
160 static void sum_val(uint64_t *dst, uint64_t *src)
169 static void pthread_double_unlock(pthread_mutex_t *lock1,
170 pthread_mutex_t *lock2)
173 pthread_mutex_unlock(lock1);
174 pthread_mutex_unlock(lock2);
178 static void pthread_double_lock(pthread_mutex_t *lock1, pthread_mutex_t *lock2)
182 pthread_mutex_lock(lock1);
183 pthread_mutex_lock(lock2);
185 pthread_mutex_lock(lock2);
186 pthread_mutex_lock(lock1);
191 static void sum_ddir(struct thread_data *dst, struct thread_data *src,
194 pthread_double_lock(&dst->io_wq.stat_lock, &src->io_wq.stat_lock);
196 sum_val(&dst->io_bytes[ddir], &src->io_bytes[ddir]);
197 sum_val(&dst->io_blocks[ddir], &src->io_blocks[ddir]);
198 sum_val(&dst->this_io_blocks[ddir], &src->this_io_blocks[ddir]);
199 sum_val(&dst->this_io_bytes[ddir], &src->this_io_bytes[ddir]);
200 sum_val(&dst->bytes_done[ddir], &src->bytes_done[ddir]);
202 pthread_double_unlock(&dst->io_wq.stat_lock, &src->io_wq.stat_lock);
205 static void io_workqueue_update_acct_fn(struct submit_worker *sw)
207 struct thread_data *src = sw->private;
208 struct thread_data *dst = sw->wq->td;
211 sum_ddir(dst, src, DDIR_READ);
213 sum_ddir(dst, src, DDIR_WRITE);
215 sum_ddir(dst, src, DDIR_TRIM);
219 struct workqueue_ops rated_wq_ops = {
220 .fn = io_workqueue_fn,
221 .pre_sleep_flush_fn = io_workqueue_pre_sleep_flush_fn,
222 .pre_sleep_fn = io_workqueue_pre_sleep_fn,
223 .update_acct_fn = io_workqueue_update_acct_fn,
224 .alloc_worker_fn = io_workqueue_alloc_fn,
225 .free_worker_fn = io_workqueue_free_fn,
226 .init_worker_fn = io_workqueue_init_worker_fn,
227 .exit_worker_fn = io_workqueue_exit_worker_fn,