engines/ime: various code and style cleanups
[fio.git] / engines / ime.c
1 /*
2  * FIO engines for DDN's Infinite Memory Engine.
3  * This file defines 3 engines: ime_psync, ime_psyncv, and ime_aio
4  *
5  * Copyright (C) 2018      DataDirect Networks. All rights reserved.
6  *
7  * This program is free software; you can redistribute it and/or
8  * modify it under the terms of the GNU General Public License,
9  * version 2 as published by the Free Software Foundation..
10  *
11  * This program is distributed in the hope that it will be useful,
12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14  * GNU General Public License for more details.
15  */
16
17 /*
18  * Some details about the new engines are given below:
19  *
20  *
21  * ime_psync:
22  * Most basic engine that issues calls to ime_native whenever an IO is queued.
23  *
24  * ime_psyncv:
25  * This engine tries to queue the IOs (by creating iovecs) if asked by FIO (via
26  * iodepth_batch). It refuses to queue when the iovecs can't be appended, and
27  * waits for FIO to issue a commit. After a call to commit and get_events, new
28  * IOs can be queued.
29  *
30  * ime_aio:
31  * This engine tries to queue the IOs (by creating iovecs) if asked by FIO (via
32  * iodepth_batch). When the iovecs can't be appended to the current request, a
33  * new request for IME is created. These requests will be issued to IME when
34  * commit is called. Contrary to ime_psyncv, there can be several requests at
35  * once. We don't need to wait for a request to terminate before creating a new
36  * one.
37  */
38
39 #include <stdio.h>
40 #include <stdlib.h>
41 #include <errno.h>
42 #include <linux/limits.h>
43 #include <ime_native.h>
44
45 #include "../fio.h"
46
47
48 /**************************************************************
49  *              Types and constants definitions
50  *
51  **************************************************************/
52
53 /* define constants for async IOs */
54 #define FIO_IME_IN_PROGRESS -1
55 #define FIO_IME_REQ_ERROR   -2
56
57 /* This flag is used when some jobs were created using threads. In that
58    case, IME can't be finalized in the engine-specific cleanup function,
59    because other threads might still use IME. Instead, IME is finalized
60    in the destructor (see fio_ime_unregister), only when the flag
61    fio_ime_is_initialized is true (which means at least one thread has
62    initialized IME). */
63 static bool fio_ime_is_initialized = false;
64
65 struct imesio_req {
66         int                     fd;             /* File descriptor */
67         enum fio_ddir   ddir;   /* Type of IO (read or write) */
68         off_t                   offset; /* File offset */
69 };
70 struct imeaio_req {
71         struct ime_aiocb        iocb;                   /* IME aio request */
72         ssize_t                 status;                 /* Status of the IME request */
73         enum fio_ddir           ddir;                   /* Type of IO (read or write) */
74         pthread_cond_t          cond_endio;             /* Condition var to notify FIO */
75         pthread_mutex_t         status_mutex;   /* Mutex for cond_endio */
76 };
77
78 /* This structure will be used for 2 engines: ime_psyncv and ime_aio */
79 struct ime_data {
80         union {
81                 struct imeaio_req       *aioreqs;       /* array of aio requests */
82                 struct imesio_req       *sioreq;        /* pointer to the only syncio request */
83         };
84         struct iovec    *iovecs;                /* array of queued iovecs */
85         struct io_u     **io_us;                /* array of queued io_u pointers */
86         struct io_u     **event_io_us;  /* array of the events retieved afer get_events*/
87         unsigned int    queued;                 /* iovecs/io_us in the queue */
88         unsigned int    events;                 /* number of committed iovecs/io_us */
89
90         /* variables used to implement a "ring" queue */
91         unsigned int depth;                     /* max entries in the queue */
92         unsigned int head;                      /* index used to append */
93         unsigned int tail;                      /* index used to pop */
94         unsigned int cur_commit;        /* index of the first uncommitted req */
95
96         /* offset used by the last iovec (used to check if the iovecs can be appended)*/
97         unsigned long long      last_offset;
98
99         /* The variables below are used for aio only */
100         struct imeaio_req       *last_req; /* last request awaiting committing */
101 };
102
103
104 /**************************************************************
105  *         Private functions for queueing/unqueueing
106  *
107  **************************************************************/
108
109 static void fio_ime_queue_incr (struct ime_data *ime_d)
110 {
111         ime_d->head = (ime_d->head + 1) % ime_d->depth;
112         ime_d->queued++;
113 }
114
115 static void fio_ime_queue_red (struct ime_data *ime_d)
116 {
117         ime_d->tail = (ime_d->tail + 1) % ime_d->depth;
118         ime_d->queued--;
119         ime_d->events--;
120 }
121
122 static void fio_ime_queue_commit (struct ime_data *ime_d, int iovcnt)
123 {
124         ime_d->cur_commit = (ime_d->cur_commit + iovcnt) % ime_d->depth;
125         ime_d->events += iovcnt;
126 }
127
128 static void fio_ime_queue_reset (struct ime_data *ime_d)
129 {
130         ime_d->head = 0;
131         ime_d->tail = 0;
132         ime_d->cur_commit = 0;
133         ime_d->queued = 0;
134         ime_d->events = 0;
135 }
136
137 /**************************************************************
138  *                   General IME functions
139  *             (needed for both sync and async IOs)
140  **************************************************************/
141
142 static char *fio_set_ime_filename(char* filename)
143 {
144         static __thread char ime_filename[PATH_MAX];
145         int ret;
146
147         ret = snprintf(ime_filename, PATH_MAX, "%s%s", DEFAULT_IME_FILE_PREFIX, filename);
148         if (ret < PATH_MAX)
149                 return ime_filename;
150
151         return NULL;
152 }
153
154 static int fio_ime_get_file_size(struct thread_data *td, struct fio_file *f)
155 {
156         struct stat buf;
157         int ret;
158         char *ime_filename;
159
160         dprint(FD_FILE, "get file size %s\n", f->file_name);
161
162         ime_filename = fio_set_ime_filename(f->file_name);
163         if (ime_filename == NULL)
164                 return 1;
165         ret = ime_native_stat(ime_filename, &buf);
166         if (ret == -1) {
167                 td_verror(td, errno, "fstat");
168                 return 1;
169         }
170
171         f->real_file_size = buf.st_size;
172         return 0;
173 }
174
175 /* This functions mimics the generic_file_open function, but issues
176    IME native calls instead of POSIX calls. */
177 static int fio_ime_open_file(struct thread_data *td, struct fio_file *f)
178 {
179         int flags = 0;
180         int ret;
181         uint64_t desired_fs;
182         char *ime_filename;
183
184         dprint(FD_FILE, "fd open %s\n", f->file_name);
185
186         if (td_trim(td)) {
187                 td_verror(td, EINVAL, "IME does not support TRIM operation");
188                 return 1;
189         }
190
191         if (td->o.oatomic) {
192                 td_verror(td, EINVAL, "IME does not support atomic IO");
193                 return 1;
194         }
195         if (td->o.odirect)
196                 flags |= O_DIRECT;
197         if (td->o.sync_io)
198                 flags |= O_SYNC;
199         if (td->o.create_on_open && td->o.allow_create)
200                 flags |= O_CREAT;
201
202         if (td_write(td)) {
203                 if (!read_only)
204                         flags |= O_RDWR;
205
206                 if (td->o.allow_create)
207                         flags |= O_CREAT;
208         } else if (td_read(td)) {
209                 flags |= O_RDONLY;
210         } else {
211                 /* We should never go here. */
212                 td_verror(td, EINVAL, "Unsopported open mode");
213                 return 1;
214         }
215
216         ime_filename = fio_set_ime_filename(f->file_name);
217         if (ime_filename == NULL)
218                 return 1;
219         f->fd = ime_native_open(ime_filename, flags, 0600);
220         if (f->fd == -1) {
221                 char buf[FIO_VERROR_SIZE];
222                 int __e = errno;
223
224                 snprintf(buf, sizeof(buf), "open(%s)", f->file_name);
225                 td_verror(td, __e, buf);
226                 return 1;
227         }
228
229         /* Now we need to make sure the real file size is sufficient for FIO
230            to do its things. This is normally done before the file open function
231            is called, but because FIO would use POSIX calls, we need to do it
232            ourselves */
233         ret = fio_ime_get_file_size(td, f);
234         if (ret < 0) {
235                 ime_native_close(f->fd);
236                 td_verror(td, errno, "ime_get_file_size");
237                 return 1;
238         }
239
240         desired_fs = f->io_size + f->file_offset;
241         if (td_write(td)) {
242                 dprint(FD_FILE, "Laying out file %s%s\n",
243                         DEFAULT_IME_FILE_PREFIX, f->file_name);
244                 if (!td->o.create_on_open &&
245                                 f->real_file_size < desired_fs &&
246                                 ime_native_ftruncate(f->fd, desired_fs) < 0) {
247                         ime_native_close(f->fd);
248                         td_verror(td, errno, "ime_native_ftruncate");
249                         return 1;
250                 }
251                 if (f->real_file_size < desired_fs)
252                         f->real_file_size = desired_fs;
253         } else if (td_read(td) && f->real_file_size < desired_fs) {
254                 ime_native_close(f->fd);
255                 log_err("error: can't read %lu bytes from file with "
256                                                 "%lu bytes\n", desired_fs, f->real_file_size);
257                 return 1;
258         }
259
260         return 0;
261 }
262
263 static int fio_ime_close_file(struct thread_data fio_unused *td, struct fio_file *f)
264 {
265         int ret = 0;
266
267         dprint(FD_FILE, "fd close %s\n", f->file_name);
268
269         if (ime_native_close(f->fd) < 0)
270                 ret = errno;
271
272         f->fd = -1;
273         return ret;
274 }
275
276 static int fio_ime_unlink_file(struct thread_data *td, struct fio_file *f)
277 {
278         char *ime_filename = fio_set_ime_filename(f->file_name);
279         int ret;
280
281         if (ime_filename == NULL)
282                 return 1;
283
284         ret = unlink(ime_filename);
285         return ret < 0 ? errno : 0;
286 }
287
288 static struct io_u *fio_ime_event(struct thread_data *td, int event)
289 {
290         struct ime_data *ime_d = td->io_ops_data;
291
292         return ime_d->event_io_us[event];
293 }
294
295 /* Setup file used to replace get_file_sizes when settin up the file.
296    Instead we will set real_file_sie to 0 for each file. This way we
297    can avoid calling ime_native_init before the forks are created. */
298 static int fio_ime_setup(struct thread_data *td)
299 {
300         struct fio_file *f;
301         unsigned int i;
302
303         for_each_file(td, f, i) {
304                 dprint(FD_FILE, "setup: set file size to 0 for %p/%d/%s\n",
305                         f, i, f->file_name);
306                 f->real_file_size = 0;
307         }
308
309         return 0;
310 }
311
312 static int fio_ime_engine_init(struct thread_data *td)
313 {
314         struct fio_file *f;
315         unsigned int i;
316
317         dprint(FD_IO, "ime engine init\n");
318         if (fio_ime_is_initialized && !td->o.use_thread) {
319                 log_err("Warning: something might go wrong. Not all threads/forks were"
320                                 " created before the FIO jobs were initialized.\n");
321         }
322
323         ime_native_init();
324         fio_ime_is_initialized = true;
325
326         /* We have to temporarily set real_file_size so that
327            FIO can initialize properly. It will be corrected
328            on file open. */
329         for_each_file(td, f, i)
330                 f->real_file_size = f->io_size + f->file_offset;
331
332         return 0;
333 }
334
335 static void fio_ime_engine_finalize(struct thread_data *td)
336 {
337         /* Only finalize IME when using forks */
338         if (!td->o.use_thread) {
339                 if (ime_native_finalize() < 0)
340                         log_err("error in ime_native_finalize\n");
341                 fio_ime_is_initialized = false;
342         }
343 }
344
345
346 /**************************************************************
347  *             Private functions for blocking IOs
348  *                     (without iovecs)
349  **************************************************************/
350
351 /* Notice: this function comes from the sync engine */
352 /* It is used by the commit function to return a proper code and fill
353    some attributes in the io_u used for the IO. */
354 static int fio_ime_psync_end(struct thread_data *td, struct io_u *io_u, ssize_t ret)
355 {
356         if (ret != (ssize_t) io_u->xfer_buflen) {
357                 if (ret >= 0) {
358                         io_u->resid = io_u->xfer_buflen - ret;
359                         io_u->error = 0;
360                         return FIO_Q_COMPLETED;
361                 } else
362                         io_u->error = errno;
363         }
364
365         if (io_u->error) {
366                 io_u_log_error(td, io_u);
367                 td_verror(td, io_u->error, "xfer");
368         }
369
370         return FIO_Q_COMPLETED;
371 }
372
373 static enum fio_q_status fio_ime_psync_queue(struct thread_data *td,
374                                            struct io_u *io_u)
375 {
376         struct fio_file *f = io_u->file;
377         ssize_t ret;
378
379         fio_ro_check(td, io_u);
380
381         if (io_u->ddir == DDIR_READ)
382                 ret = ime_native_pread(f->fd, io_u->xfer_buf, io_u->xfer_buflen, io_u->offset);
383         else if (io_u->ddir == DDIR_WRITE)
384                 ret = ime_native_pwrite(f->fd, io_u->xfer_buf, io_u->xfer_buflen, io_u->offset);
385         else if (io_u->ddir == DDIR_SYNC)
386                 ret = ime_native_fsync(f->fd);
387         else {
388                 ret = io_u->xfer_buflen;
389                 io_u->error = EINVAL;
390         }
391
392         return fio_ime_psync_end(td, io_u, ret);
393 }
394
395
396 /**************************************************************
397  *             Private functions for blocking IOs
398  *                       (with iovecs)
399  **************************************************************/
400
401 static bool fio_ime_psyncv_can_queue(struct ime_data *ime_d, struct io_u *io_u)
402 {
403         /* We can only queue if:
404           - There are no queued iovecs
405           - Or if there is at least one:
406                  - There must be no event waiting for retrieval
407                  - The offsets must be contiguous
408                  - The ddir and fd must be the same */
409         return (ime_d->queued == 0 || (
410                         ime_d->events == 0 &&
411                         ime_d->last_offset == io_u->offset &&
412                         ime_d->sioreq->ddir == io_u->ddir &&
413                         ime_d->sioreq->fd == io_u->file->fd));
414 }
415
416 /* Before using this function, we should have already
417    ensured that the queue is not full */
418 static void fio_ime_psyncv_enqueue(struct ime_data *ime_d, struct io_u *io_u)
419 {
420         struct imesio_req *ioreq = ime_d->sioreq;
421         struct iovec *iov = &ime_d->iovecs[ime_d->head];
422
423         iov->iov_base = io_u->xfer_buf;
424         iov->iov_len = io_u->xfer_buflen;
425
426         if (ime_d->queued == 0) {
427                 ioreq->offset = io_u->offset;
428                 ioreq->ddir = io_u->ddir;
429                 ioreq->fd = io_u->file->fd;
430         }
431
432         ime_d->io_us[ime_d->head] = io_u;
433         ime_d->last_offset = io_u->offset + io_u->xfer_buflen;
434         fio_ime_queue_incr(ime_d);
435 }
436
437 /* Tries to queue an IO. It will fail if the IO can't be appended to the
438    current request or if the current request has been committed but not
439    yet retrieved by get_events. */
440 static enum fio_q_status fio_ime_psyncv_queue(struct thread_data *td,
441         struct io_u *io_u)
442 {
443         struct ime_data *ime_d = td->io_ops_data;
444
445         fio_ro_check(td, io_u);
446
447         if (ime_d->queued == ime_d->depth)
448                 return FIO_Q_BUSY;
449
450         if (io_u->ddir == DDIR_READ || io_u->ddir == DDIR_WRITE) {
451                 if (!fio_ime_psyncv_can_queue(ime_d, io_u))
452                         return FIO_Q_BUSY;
453
454                 dprint(FD_IO, "queue: ddir=%d at %u commit=%u queued=%u events=%u\n",
455                         io_u->ddir, ime_d->head, ime_d->cur_commit,
456                         ime_d->queued, ime_d->events);
457                 fio_ime_psyncv_enqueue(ime_d, io_u);
458                 return FIO_Q_QUEUED;
459         } else if (io_u->ddir == DDIR_SYNC) {
460                 if (ime_native_fsync(io_u->file->fd) < 0) {
461                         io_u->error = errno;
462                         td_verror(td, io_u->error, "fsync");
463                 }
464                 return FIO_Q_COMPLETED;
465         } else {
466                 io_u->error = EINVAL;
467                 td_verror(td, io_u->error, "wrong ddir");
468                 return FIO_Q_COMPLETED;
469         }
470 }
471
472 /* Notice: this function comes from the sync engine */
473 /* It is used by the commit function to return a proper code and fill
474    some attributes in the io_us appended to the current request. */
475 static int fio_ime_psyncv_end(struct thread_data *td, ssize_t bytes)
476 {
477         struct ime_data *ime_d = td->io_ops_data;
478         struct io_u *io_u;
479         unsigned int i;
480         int err = errno;
481
482         for (i = 0; i < ime_d->queued; i++) {
483                 io_u = ime_d->io_us[i];
484
485                 if (bytes == -1)
486                         io_u->error = err;
487                 else {
488                         unsigned int this_io;
489
490                         this_io = bytes;
491                         if (this_io > io_u->xfer_buflen)
492                                 this_io = io_u->xfer_buflen;
493
494                         io_u->resid = io_u->xfer_buflen - this_io;
495                         io_u->error = 0;
496                         bytes -= this_io;
497                 }
498         }
499
500         if (bytes == -1) {
501                 td_verror(td, err, "xfer psyncv");
502                 return -err;
503         }
504
505         return 0;
506 }
507
508 /* Commits the current request by calling ime_native (with one or several
509    iovecs). After this commit, the corresponding events (one per iovec)
510    can be retrieved by get_events. */
511 static int fio_ime_psyncv_commit(struct thread_data *td)
512 {
513         struct ime_data *ime_d = td->io_ops_data;
514         struct imesio_req *ioreq;
515         int ret = 0;
516
517         /* Exit if there are no (new) events to commit
518            or if the previous committed event haven't been retrieved */
519         if (!ime_d->queued || ime_d->events)
520                 return 0;
521
522         ioreq = ime_d->sioreq;
523         ime_d->events = ime_d->queued;
524         if (ioreq->ddir == DDIR_READ)
525                 ret = ime_native_preadv(ioreq->fd, ime_d->iovecs, ime_d->queued, ioreq->offset);
526         else
527                 ret = ime_native_pwritev(ioreq->fd, ime_d->iovecs, ime_d->queued, ioreq->offset);
528
529         dprint(FD_IO, "committed %d iovecs\n", ime_d->queued);
530
531         return fio_ime_psyncv_end(td, ret);
532 }
533
534 static int fio_ime_psyncv_getevents(struct thread_data *td, unsigned int min,
535                                 unsigned int max, const struct timespec *t)
536 {
537         struct ime_data *ime_d = td->io_ops_data;
538         struct io_u *io_u;
539         int events = 0;
540         unsigned int count;
541
542         if (ime_d->events) {
543                 for (count = 0; count < ime_d->events; count++) {
544                         io_u = ime_d->io_us[count];
545                         ime_d->event_io_us[events] = io_u;
546                         events++;
547                 }
548                 fio_ime_queue_reset(ime_d);
549         }
550
551         dprint(FD_IO, "getevents(%u,%u) ret=%d queued=%u events=%u\n",
552                 min, max, events, ime_d->queued, ime_d->events);
553         return events;
554 }
555
556 static int fio_ime_psyncv_init(struct thread_data *td)
557 {
558         struct ime_data *ime_d;
559
560         if (fio_ime_engine_init(td) < 0)
561                 return 1;
562
563         ime_d = calloc(1, sizeof(*ime_d));
564
565         ime_d->sioreq = malloc(sizeof(struct imesio_req));
566         ime_d->iovecs = malloc(td->o.iodepth * sizeof(struct iovec));
567         ime_d->io_us = malloc(2 * td->o.iodepth * sizeof(struct io_u *));
568         ime_d->event_io_us = ime_d->io_us + td->o.iodepth;
569
570         ime_d->depth = td->o.iodepth;
571
572         td->io_ops_data = ime_d;
573         return 0;
574 }
575
576 static void fio_ime_psyncv_clean(struct thread_data *td)
577 {
578         struct ime_data *ime_d = td->io_ops_data;
579
580         if (ime_d) {
581                 free(ime_d->sioreq);
582                 free(ime_d->iovecs);
583                 free(ime_d->io_us);
584                 free(ime_d);
585                 td->io_ops_data = NULL;
586         }
587
588         fio_ime_engine_finalize(td);
589 }
590
591
592 /**************************************************************
593  *           Private functions for non-blocking IOs
594  *
595  **************************************************************/
596
597 void fio_ime_aio_complete_cb  (struct ime_aiocb *aiocb, int err,
598                                                            ssize_t bytes)
599 {
600         struct imeaio_req *ioreq = (struct imeaio_req *) aiocb->user_context;
601
602         pthread_mutex_lock(&ioreq->status_mutex);
603         ioreq->status = err == 0 ? bytes : FIO_IME_REQ_ERROR;
604         pthread_mutex_unlock(&ioreq->status_mutex);
605
606         pthread_cond_signal(&ioreq->cond_endio);
607 }
608
609 static bool fio_ime_aio_can_queue (struct ime_data *ime_d, struct io_u *io_u)
610 {
611         /* So far we can queue in any case. */
612         return true;
613 }
614 static bool fio_ime_aio_can_append (struct ime_data *ime_d, struct io_u *io_u)
615 {
616         /* We can only append if:
617                 - The iovecs will be contiguous in the array
618                 - There is already a queued iovec
619                 - The offsets are contiguous
620                 - The ddir and fs are the same */
621         return (ime_d->head != 0 &&
622                         ime_d->queued - ime_d->events > 0 &&
623                         ime_d->last_offset == io_u->offset &&
624                         ime_d->last_req->ddir == io_u->ddir &&
625                         ime_d->last_req->iocb.fd == io_u->file->fd);
626 }
627
628 /* Before using this function, we should have already
629    ensured that the queue is not full */
630 static void fio_ime_aio_enqueue(struct ime_data *ime_d, struct io_u *io_u)
631 {
632         struct imeaio_req *ioreq = &ime_d->aioreqs[ime_d->head];
633         struct ime_aiocb *iocb = &ioreq->iocb;
634         struct iovec *iov = &ime_d->iovecs[ime_d->head];
635
636         iov->iov_base = io_u->xfer_buf;
637         iov->iov_len = io_u->xfer_buflen;
638
639         if (fio_ime_aio_can_append(ime_d, io_u))
640                 ime_d->last_req->iocb.iovcnt++;
641         else {
642                 ioreq->status = FIO_IME_IN_PROGRESS;
643                 ioreq->ddir = io_u->ddir;
644                 ime_d->last_req = ioreq;
645
646                 iocb->complete_cb = &fio_ime_aio_complete_cb;
647                 iocb->fd = io_u->file->fd;
648                 iocb->file_offset = io_u->offset;
649                 iocb->iov = iov;
650                 iocb->iovcnt = 1;
651                 iocb->flags = 0;
652                 iocb->user_context = (intptr_t) ioreq;
653         }
654
655         ime_d->io_us[ime_d->head] = io_u;
656         ime_d->last_offset = io_u->offset + io_u->xfer_buflen;
657         fio_ime_queue_incr(ime_d);
658 }
659
660 /* Tries to queue an IO. It will create a new request if the IO can't be
661    appended to the current request. It will fail if the queue can't contain
662    any more io_u/iovec. In this case, commit and then get_events need to be
663    called. */
664 static enum fio_q_status fio_ime_aio_queue(struct thread_data *td,
665                 struct io_u *io_u)
666 {
667         struct ime_data *ime_d = td->io_ops_data;
668
669         fio_ro_check(td, io_u);
670
671         dprint(FD_IO, "queue: ddir=%d at %u commit=%u queued=%u events=%u\n",
672                 io_u->ddir, ime_d->head, ime_d->cur_commit,
673                 ime_d->queued, ime_d->events);
674
675         if (ime_d->queued == ime_d->depth)
676                 return FIO_Q_BUSY;
677
678         if (io_u->ddir == DDIR_READ || io_u->ddir == DDIR_WRITE) {
679                 if (!fio_ime_aio_can_queue(ime_d, io_u))
680                         return FIO_Q_BUSY;
681
682                 fio_ime_aio_enqueue(ime_d, io_u);
683                 return FIO_Q_QUEUED;
684         } else if (io_u->ddir == DDIR_SYNC) {
685                 if (ime_native_fsync(io_u->file->fd) < 0) {
686                         io_u->error = errno;
687                         td_verror(td, io_u->error, "fsync");
688                 }
689                 return FIO_Q_COMPLETED;
690         } else {
691                 io_u->error = EINVAL;
692                 td_verror(td, io_u->error, "wrong ddir");
693                 return FIO_Q_COMPLETED;
694         }
695 }
696
697 static int fio_ime_aio_commit(struct thread_data *td)
698 {
699         struct ime_data *ime_d = td->io_ops_data;
700         struct imeaio_req *ioreq;
701         int ret = 0;
702
703         /* Loop while there are events to commit */
704         while (ime_d->queued - ime_d->events) {
705                 ioreq = &ime_d->aioreqs[ime_d->cur_commit];
706                 if (ioreq->ddir == DDIR_READ)
707                         ret = ime_native_aio_read(&ioreq->iocb);
708                 else
709                         ret = ime_native_aio_write(&ioreq->iocb);
710
711                 fio_ime_queue_commit(ime_d, ioreq->iocb.iovcnt);
712
713                 /* fio needs a negative error code */
714                 if (ret < 0) {
715                         ioreq->status = FIO_IME_REQ_ERROR;
716                         return -errno;
717                 }
718
719                 io_u_mark_submit(td, ioreq->iocb.iovcnt);
720                 dprint(FD_IO, "committed %d iovecs commit=%u queued=%u events=%u\n",
721                         ioreq->iocb.iovcnt, ime_d->cur_commit,
722                         ime_d->queued, ime_d->events);
723         }
724
725         return 0;
726 }
727
728 static int fio_ime_aio_getevents(struct thread_data *td, unsigned int min,
729                                 unsigned int max, const struct timespec *t)
730 {
731         struct ime_data *ime_d = td->io_ops_data;
732         struct imeaio_req *ioreq;
733         struct io_u *io_u;
734         int events = 0;
735         unsigned int count;
736         ssize_t bytes;
737
738         while (ime_d->events) {
739                 ioreq = &ime_d->aioreqs[ime_d->tail];
740
741                 /* Break if we already got events, and if we will
742                    exceed max if we append the next events */
743                 if (events && events + ioreq->iocb.iovcnt > max)
744                         break;
745
746                 if (ioreq->status != FIO_IME_IN_PROGRESS) {
747
748                         bytes = ioreq->status;
749                         for (count = 0; count < ioreq->iocb.iovcnt; count++) {
750                                 io_u = ime_d->io_us[ime_d->tail];
751                                 ime_d->event_io_us[events] = io_u;
752                                 events++;
753                                 fio_ime_queue_red(ime_d);
754
755                                 if (ioreq->status == FIO_IME_REQ_ERROR)
756                                         io_u->error = EIO;
757                                 else {
758                                         io_u->resid = bytes > io_u->xfer_buflen ?
759                                                                         0 : io_u->xfer_buflen - bytes;
760                                         io_u->error = 0;
761                                         bytes -= io_u->xfer_buflen - io_u->resid;
762                                 }
763                         }
764                 } else {
765                         pthread_mutex_lock(&ioreq->status_mutex);
766                         while (ioreq->status == FIO_IME_IN_PROGRESS)
767                                 pthread_cond_wait(&ioreq->cond_endio, &ioreq->status_mutex);
768                         pthread_mutex_unlock(&ioreq->status_mutex);
769                 }
770
771         }
772
773         dprint(FD_IO, "getevents(%u,%u) ret=%d queued=%u events=%u\n", min, max,
774                 events, ime_d->queued, ime_d->events);
775         return events;
776 }
777
778 static int fio_ime_aio_init(struct thread_data *td)
779 {
780         struct ime_data *ime_d;
781         struct imeaio_req *ioreq;
782         unsigned int i;
783
784         if (fio_ime_engine_init(td) < 0)
785                 return 1;
786
787         ime_d = calloc(1, sizeof(*ime_d));
788
789         ime_d->aioreqs = malloc(td->o.iodepth * sizeof(struct imeaio_req));
790         ime_d->iovecs = malloc(td->o.iodepth * sizeof(struct iovec));
791         ime_d->io_us = malloc(2 * td->o.iodepth * sizeof(struct io_u *));
792         ime_d->event_io_us = ime_d->io_us + td->o.iodepth;
793
794         ime_d->depth = td->o.iodepth;
795         for (i = 0; i < ime_d->depth; i++) {
796                 ioreq = &ime_d->aioreqs[i];
797                 pthread_cond_init(&ioreq->cond_endio, NULL);
798                 pthread_mutex_init(&ioreq->status_mutex, NULL);
799         }
800
801         td->io_ops_data = ime_d;
802         return 0;
803 }
804
805 static void fio_ime_aio_clean(struct thread_data *td)
806 {
807         struct ime_data *ime_d = td->io_ops_data;
808         struct imeaio_req *ioreq;
809         unsigned int i;
810
811         if (ime_d) {
812                 for (i = 0; i < ime_d->depth; i++) {
813                         ioreq = &ime_d->aioreqs[i];
814                         pthread_cond_destroy(&ioreq->cond_endio);
815                         pthread_mutex_destroy(&ioreq->status_mutex);
816                 }
817                 free(ime_d->aioreqs);
818                 free(ime_d->iovecs);
819                 free(ime_d->io_us);
820                 free(ime_d);
821                 td->io_ops_data = NULL;
822         }
823
824         fio_ime_engine_finalize(td);
825 }
826
827
828 /**************************************************************
829  *                   IO engines definitions
830  *
831  **************************************************************/
832
833 /* The FIO_DISKLESSIO flag used for these engines is necessary to prevent
834    FIO from using POSIX calls. See fio_ime_open_file for more details. */
835
836 static struct ioengine_ops ioengine_prw = {
837         .name           = "ime_psync",
838         .version        = FIO_IOOPS_VERSION,
839         .setup          = fio_ime_setup,
840         .init           = fio_ime_engine_init,
841         .cleanup        = fio_ime_engine_finalize,
842         .queue          = fio_ime_psync_queue,
843         .open_file      = fio_ime_open_file,
844         .close_file     = fio_ime_close_file,
845         .get_file_size  = fio_ime_get_file_size,
846         .unlink_file    = fio_ime_unlink_file,
847         .flags          = FIO_SYNCIO | FIO_DISKLESSIO,
848 };
849
850 static struct ioengine_ops ioengine_pvrw = {
851         .name           = "ime_psyncv",
852         .version        = FIO_IOOPS_VERSION,
853         .setup          = fio_ime_setup,
854         .init           = fio_ime_psyncv_init,
855         .cleanup        = fio_ime_psyncv_clean,
856         .queue          = fio_ime_psyncv_queue,
857         .commit         = fio_ime_psyncv_commit,
858         .getevents      = fio_ime_psyncv_getevents,
859         .event          = fio_ime_event,
860         .open_file      = fio_ime_open_file,
861         .close_file     = fio_ime_close_file,
862         .get_file_size  = fio_ime_get_file_size,
863         .unlink_file    = fio_ime_unlink_file,
864         .flags          = FIO_SYNCIO | FIO_DISKLESSIO,
865 };
866
867 static struct ioengine_ops ioengine_aio = {
868         .name           = "ime_aio",
869         .version        = FIO_IOOPS_VERSION,
870         .setup          = fio_ime_setup,
871         .init           = fio_ime_aio_init,
872         .cleanup        = fio_ime_aio_clean,
873         .queue          = fio_ime_aio_queue,
874         .commit         = fio_ime_aio_commit,
875         .getevents      = fio_ime_aio_getevents,
876         .event          = fio_ime_event,
877         .open_file      = fio_ime_open_file,
878         .close_file     = fio_ime_close_file,
879         .get_file_size  = fio_ime_get_file_size,
880         .unlink_file    = fio_ime_unlink_file,
881         .flags          = FIO_DISKLESSIO,
882 };
883
884 static void fio_init fio_ime_register(void)
885 {
886         register_ioengine(&ioengine_prw);
887         register_ioengine(&ioengine_pvrw);
888         register_ioengine(&ioengine_aio);
889 }
890
891 static void fio_exit fio_ime_unregister(void)
892 {
893         unregister_ioengine(&ioengine_prw);
894         unregister_ioengine(&ioengine_pvrw);
895         unregister_ioengine(&ioengine_aio);
896
897         if (fio_ime_is_initialized && ime_native_finalize() < 0)
898                 log_err("Warning: IME did not finalize properly\n");
899 }