3 #include "lib/getrusage.h"
5 static int 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);
55 static bool io_workqueue_pre_sleep_flush_fn(struct submit_worker *sw)
57 struct thread_data *td = sw->private;
59 if (td->io_u_queued || td->cur_depth || td->io_u_in_flight)
65 static void io_workqueue_pre_sleep_fn(struct submit_worker *sw)
67 struct thread_data *td = sw->private;
70 ret = io_u_quiesce(td);
75 static int io_workqueue_alloc_fn(struct submit_worker *sw)
77 struct thread_data *td;
79 td = calloc(1, sizeof(*td));
84 static void io_workqueue_free_fn(struct submit_worker *sw)
90 static int io_workqueue_init_worker_fn(struct submit_worker *sw)
92 struct thread_data *parent = sw->wq->td;
93 struct thread_data *td = sw->private;
96 memcpy(&td->o, &parent->o, sizeof(td->o));
97 memcpy(&td->ts, &parent->ts, sizeof(td->ts));
98 td->o.uid = td->o.gid = -1U;
99 dup_files(td, parent);
101 fio_options_mem_dupe(td);
103 if (ioengine_load(td))
107 td->io_ops->flags |= FIO_RAWIO;
111 INIT_FLIST_HEAD(&td->io_log_list);
112 INIT_FLIST_HEAD(&td->io_hist_list);
113 INIT_FLIST_HEAD(&td->verify_list);
114 INIT_FLIST_HEAD(&td->trim_list);
115 INIT_FLIST_HEAD(&td->next_rand_list);
116 td->io_hist_tree = RB_ROOT;
122 fio_gettime(&td->epoch, NULL);
123 fio_getrusage(&td->ru_start);
124 clear_io_state(td, 1);
126 td_set_runstate(td, TD_RUNNING);
127 td->flags |= TD_F_CHILD;
138 static void io_workqueue_exit_worker_fn(struct submit_worker *sw,
139 unsigned int *sum_cnt)
141 struct thread_data *td = sw->private;
144 sum_thread_stats(&sw->wq->td->ts, &td->ts, *sum_cnt == 1);
146 fio_options_free(td);
147 close_and_free_files(td);
150 td_set_runstate(td, TD_EXITED);
154 static void sum_val(uint64_t *dst, uint64_t *src)
157 __sync_fetch_and_add(dst, *src);
162 static void sum_val(uint64_t *dst, uint64_t *src)
171 static void pthread_double_unlock(pthread_mutex_t *lock1,
172 pthread_mutex_t *lock2)
175 pthread_mutex_unlock(lock1);
176 pthread_mutex_unlock(lock2);
180 static void pthread_double_lock(pthread_mutex_t *lock1, pthread_mutex_t *lock2)
184 pthread_mutex_lock(lock1);
185 pthread_mutex_lock(lock2);
187 pthread_mutex_lock(lock2);
188 pthread_mutex_lock(lock1);
193 static void sum_ddir(struct thread_data *dst, struct thread_data *src,
196 pthread_double_lock(&dst->io_wq.stat_lock, &src->io_wq.stat_lock);
198 sum_val(&dst->io_bytes[ddir], &src->io_bytes[ddir]);
199 sum_val(&dst->io_blocks[ddir], &src->io_blocks[ddir]);
200 sum_val(&dst->this_io_blocks[ddir], &src->this_io_blocks[ddir]);
201 sum_val(&dst->this_io_bytes[ddir], &src->this_io_bytes[ddir]);
202 sum_val(&dst->bytes_done[ddir], &src->bytes_done[ddir]);
204 pthread_double_unlock(&dst->io_wq.stat_lock, &src->io_wq.stat_lock);
207 static void io_workqueue_update_acct_fn(struct submit_worker *sw)
209 struct thread_data *src = sw->private;
210 struct thread_data *dst = sw->wq->td;
213 sum_ddir(dst, src, DDIR_READ);
215 sum_ddir(dst, src, DDIR_WRITE);
217 sum_ddir(dst, src, DDIR_TRIM);
221 struct workqueue_ops rated_wq_ops = {
222 .fn = io_workqueue_fn,
223 .pre_sleep_flush_fn = io_workqueue_pre_sleep_flush_fn,
224 .pre_sleep_fn = io_workqueue_pre_sleep_fn,
225 .update_acct_fn = io_workqueue_update_acct_fn,
226 .alloc_worker_fn = io_workqueue_alloc_fn,
227 .free_worker_fn = io_workqueue_free_fn,
228 .init_worker_fn = io_workqueue_init_worker_fn,
229 .exit_worker_fn = io_workqueue_exit_worker_fn,