Abstract out generic sync helper
[fio.git] / engines / sync.c
1 /*
2  * sync/psync engine
3  *
4  * IO engine that does regular read(2)/write(2) with lseek(2) to transfer
5  * data and IO engine that does regular pread(2)/pwrite(2) to transfer data.
6  *
7  */
8 #include <stdio.h>
9 #include <stdlib.h>
10 #include <unistd.h>
11 #include <sys/uio.h>
12 #include <errno.h>
13 #include <assert.h>
14
15 #include "../fio.h"
16
17 struct syncio_data {
18         struct iovec *iovecs;
19         struct io_u **io_us;
20         unsigned int queued;
21         unsigned int events;
22         unsigned long queued_bytes;
23
24         unsigned long long last_offset;
25         struct fio_file *last_file;
26         enum fio_ddir last_ddir;
27 };
28
29 static int fio_syncio_prep(struct thread_data *td, struct io_u *io_u)
30 {
31         struct fio_file *f = io_u->file;
32
33         if (ddir_sync(io_u->ddir))
34                 return 0;
35
36         if (f->file_pos != -1ULL && f->file_pos == io_u->offset)
37                 return 0;
38
39         if (lseek(f->fd, io_u->offset, SEEK_SET) == -1) {
40                 td_verror(td, errno, "lseek");
41                 return 1;
42         }
43
44         return 0;
45 }
46
47 static int fio_io_end(struct thread_data *td, struct io_u *io_u, int ret)
48 {
49         if (io_u->file && ret >= 0)
50                 io_u->file->file_pos = io_u->offset + ret;
51
52         if (ret != (int) io_u->xfer_buflen) {
53                 if (ret >= 0) {
54                         io_u->resid = io_u->xfer_buflen - ret;
55                         io_u->error = 0;
56                         return FIO_Q_COMPLETED;
57                 } else
58                         io_u->error = errno;
59         }
60
61         if (io_u->error)
62                 td_verror(td, io_u->error, "xfer");
63
64         return FIO_Q_COMPLETED;
65 }
66
67 static int fio_psyncio_queue(struct thread_data *td, struct io_u *io_u)
68 {
69         struct fio_file *f = io_u->file;
70         int ret;
71
72         fio_ro_check(td, io_u);
73
74         if (io_u->ddir == DDIR_READ)
75                 ret = pread(f->fd, io_u->xfer_buf, io_u->xfer_buflen, io_u->offset);
76         else if (io_u->ddir == DDIR_WRITE)
77                 ret = pwrite(f->fd, io_u->xfer_buf, io_u->xfer_buflen, io_u->offset);
78         else
79                 ret = do_io_u_sync(td, io_u);
80
81         return fio_io_end(td, io_u, ret);
82 }
83
84 static int fio_syncio_queue(struct thread_data *td, struct io_u *io_u)
85 {
86         struct fio_file *f = io_u->file;
87         int ret;
88
89         fio_ro_check(td, io_u);
90
91         if (io_u->ddir == DDIR_READ)
92                 ret = read(f->fd, io_u->xfer_buf, io_u->xfer_buflen);
93         else if (io_u->ddir == DDIR_WRITE)
94                 ret = write(f->fd, io_u->xfer_buf, io_u->xfer_buflen);
95         else
96                 ret = do_io_u_sync(td, io_u);
97
98         return fio_io_end(td, io_u, ret);
99 }
100
101 static int fio_vsyncio_getevents(struct thread_data *td, unsigned int min,
102                                  unsigned int max,
103                                  struct timespec fio_unused *t)
104 {
105         struct syncio_data *sd = td->io_ops->data;
106         int ret;
107
108         if (min) {
109                 ret = sd->events;
110                 sd->events = 0;
111         } else
112                 ret = 0;
113
114         dprint(FD_IO, "vsyncio_getevents: min=%d,max=%d: %d\n", min, max, ret);
115         return ret;
116 }
117
118 static struct io_u *fio_vsyncio_event(struct thread_data *td, int event)
119 {
120         struct syncio_data *sd = td->io_ops->data;
121
122         return sd->io_us[event];
123 }
124
125 static int fio_vsyncio_append(struct thread_data *td, struct io_u *io_u)
126 {
127         struct syncio_data *sd = td->io_ops->data;
128
129         if (ddir_sync(io_u->ddir))
130                 return 0;
131
132         if (io_u->offset == sd->last_offset && io_u->file == sd->last_file &&
133             io_u->ddir == sd->last_ddir)
134                 return 1;
135
136         return 0;
137 }
138
139 static void fio_vsyncio_set_iov(struct syncio_data *sd, struct io_u *io_u,
140                                 int index)
141 {
142         sd->io_us[index] = io_u;
143         sd->iovecs[index].iov_base = io_u->xfer_buf;
144         sd->iovecs[index].iov_len = io_u->xfer_buflen;
145         sd->last_offset = io_u->offset + io_u->xfer_buflen;
146         sd->last_file = io_u->file;
147         sd->last_ddir = io_u->ddir;
148         sd->queued_bytes += io_u->xfer_buflen;
149         sd->queued++;
150 }
151
152 static int fio_vsyncio_queue(struct thread_data *td, struct io_u *io_u)
153 {
154         struct syncio_data *sd = td->io_ops->data;
155
156         fio_ro_check(td, io_u);
157
158         if (!fio_vsyncio_append(td, io_u)) {
159                 dprint(FD_IO, "vsyncio_queue: no append (%d)\n", sd->queued);
160                 /*
161                  * If we can't append and have stuff queued, tell fio to
162                  * commit those first and then retry this io
163                  */
164                 if (sd->queued)
165                         return FIO_Q_BUSY;
166                 if (ddir_sync(io_u->ddir)) {
167                         int ret = do_io_u_sync(td, io_u);
168
169                         return fio_io_end(td, io_u, ret);
170                 }
171
172                 sd->queued = 0;
173                 sd->queued_bytes = 0;
174                 fio_vsyncio_set_iov(sd, io_u, 0);
175         } else {
176                 if (sd->queued == td->o.iodepth) {
177                         dprint(FD_IO, "vsyncio_queue: max depth %d\n", sd->queued);
178                         return FIO_Q_BUSY;
179                 }
180
181                 dprint(FD_IO, "vsyncio_queue: append\n");
182                 fio_vsyncio_set_iov(sd, io_u, sd->queued);
183         }
184
185         dprint(FD_IO, "vsyncio_queue: depth now %d\n", sd->queued);
186         return FIO_Q_QUEUED;
187 }
188
189 /*
190  * Check that we transferred all bytes, or saw an error, etc
191  */
192 static int fio_vsyncio_end(struct thread_data *td, ssize_t bytes)
193 {
194         struct syncio_data *sd = td->io_ops->data;
195         struct io_u *io_u;
196         unsigned int i;
197         int err;
198
199         /*
200          * transferred everything, perfect
201          */
202         if (bytes == sd->queued_bytes)
203                 return 0;
204
205         err = errno;
206         for (i = 0; i < sd->queued; i++) {
207                 io_u = sd->io_us[i];
208
209                 if (bytes == -1) {
210                         io_u->error = err;
211                 } else {
212                         unsigned int this_io;
213
214                         this_io = bytes;
215                         if (this_io > io_u->xfer_buflen)
216                                 this_io = io_u->xfer_buflen;
217
218                         io_u->resid = io_u->xfer_buflen - this_io;
219                         io_u->error = 0;
220                         bytes -= this_io;
221                 }
222         }
223
224         if (bytes == -1) {
225                 td_verror(td, err, "xfer vsync");
226                 return -err;
227         }
228
229         return 0;
230 }
231
232 static int fio_vsyncio_commit(struct thread_data *td)
233 {
234         struct syncio_data *sd = td->io_ops->data;
235         struct fio_file *f;
236         ssize_t ret;
237
238         if (!sd->queued)
239                 return 0;
240
241         io_u_mark_submit(td, sd->queued);
242         f = sd->last_file;
243
244         if (lseek(f->fd, sd->io_us[0]->offset, SEEK_SET) == -1) {
245                 int err = -errno;
246
247                 td_verror(td, errno, "lseek");
248                 return err;
249         }
250
251         if (sd->last_ddir == DDIR_READ)
252                 ret = readv(f->fd, sd->iovecs, sd->queued);
253         else
254                 ret = writev(f->fd, sd->iovecs, sd->queued);
255
256         dprint(FD_IO, "vsyncio_commit: %d\n", (int) ret);
257         sd->events = sd->queued;
258         sd->queued = 0;
259         return fio_vsyncio_end(td, ret);
260 }
261
262 static int fio_vsyncio_init(struct thread_data *td)
263 {
264         struct syncio_data *sd;
265
266         sd = malloc(sizeof(*sd));
267         memset(sd, 0, sizeof(*sd));
268         sd->last_offset = -1ULL;
269         sd->iovecs = malloc(td->o.iodepth * sizeof(struct iovec));
270         sd->io_us = malloc(td->o.iodepth * sizeof(struct io_u *));
271
272         td->io_ops->data = sd;
273         return 0;
274 }
275
276 static void fio_vsyncio_cleanup(struct thread_data *td)
277 {
278         struct syncio_data *sd = td->io_ops->data;
279
280         free(sd->iovecs);
281         free(sd->io_us);
282         free(sd);
283 }
284
285 static struct ioengine_ops ioengine_rw = {
286         .name           = "sync",
287         .version        = FIO_IOOPS_VERSION,
288         .prep           = fio_syncio_prep,
289         .queue          = fio_syncio_queue,
290         .open_file      = generic_open_file,
291         .close_file     = generic_close_file,
292         .get_file_size  = generic_get_file_size,
293         .flags          = FIO_SYNCIO,
294 };
295
296 static struct ioengine_ops ioengine_prw = {
297         .name           = "psync",
298         .version        = FIO_IOOPS_VERSION,
299         .queue          = fio_psyncio_queue,
300         .open_file      = generic_open_file,
301         .close_file     = generic_close_file,
302         .get_file_size  = generic_get_file_size,
303         .flags          = FIO_SYNCIO,
304 };
305
306 static struct ioengine_ops ioengine_vrw = {
307         .name           = "vsync",
308         .version        = FIO_IOOPS_VERSION,
309         .init           = fio_vsyncio_init,
310         .cleanup        = fio_vsyncio_cleanup,
311         .queue          = fio_vsyncio_queue,
312         .commit         = fio_vsyncio_commit,
313         .event          = fio_vsyncio_event,
314         .getevents      = fio_vsyncio_getevents,
315         .open_file      = generic_open_file,
316         .close_file     = generic_close_file,
317         .get_file_size  = generic_get_file_size,
318         .flags          = FIO_SYNCIO,
319 };
320
321 static void fio_init fio_syncio_register(void)
322 {
323         register_ioengine(&ioengine_rw);
324         register_ioengine(&ioengine_prw);
325         register_ioengine(&ioengine_vrw);
326 }
327
328 static void fio_exit fio_syncio_unregister(void)
329 {
330         unregister_ioengine(&ioengine_rw);
331         unregister_ioengine(&ioengine_prw);
332         unregister_ioengine(&ioengine_vrw);
333 }