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 #include "../optgroup.h"
  17
  18 /*
  19  * Sync engine uses engine_data to store last offset
  20  */
  21 #define LAST_POS(f)     ((f)->engine_pos)
  22
  23 struct syncio_data {
  24         struct iovec *iovecs;
  25         struct io_u **io_us;
  26         unsigned int queued;
  27         unsigned int events;
  28         unsigned long queued_bytes;
  29
  30         unsigned long long last_offset;
  31         struct fio_file *last_file;
  32         enum fio_ddir last_ddir;
  33 };
  34
  35 #ifdef FIO_HAVE_PWRITEV2
  36 struct psyncv2_options {
  37         void *pad;
  38         unsigned int hipri;
  39         unsigned int stream;
  40 };
  41
  42 static struct fio_option options[] = {
  43         {
  44                 .name   = "hipri",
  45                 .lname  = "RWF_HIPRI",
  46                 .type   = FIO_OPT_STR_SET,
  47                 .off1   = offsetof(struct psyncv2_options, hipri),
  48                 .help   = "Set RWF_HIPRI for pwritev2/preadv2",
  49                 .category = FIO_OPT_C_ENGINE,
  50                 .group  = FIO_OPT_G_INVALID,
  51         },
  52         {
  53                 .name   = "stream",
  54                 .lname  = "Stream ID",
  55                 .type   = FIO_OPT_STR,
  56                 .off1   = offsetof(struct psyncv2_options, stream),
  57                 .help   = "Set expected write life time",
  58                 .category = FIO_OPT_C_ENGINE,
  59                 .group  = FIO_OPT_G_INVALID,
  60                 .posval = {
  61                           { .ival = "short",
  62                             .oval = RWF_WRITE_LIFE_SHORT,
  63                           },
  64                           { .ival = "medium",
  65                             .oval = RWF_WRITE_LIFE_MEDIUM,
  66                           },
  67                           { .ival = "long",
  68                             .oval = RWF_WRITE_LIFE_LONG,
  69                           },
  70                           { .ival = "extreme",
  71                             .oval = RWF_WRITE_LIFE_EXTREME,
  72                           },
  73                 },
  74         },
  75         {
  76                 .name   = NULL,
  77         },
  78 };
  79 #endif
  80
  81 static int fio_syncio_prep(struct thread_data *td, struct io_u *io_u)
  82 {
  83         struct fio_file *f = io_u->file;
  84
  85         if (!ddir_rw(io_u->ddir))
  86                 return 0;
  87
  88         if (LAST_POS(f) != -1ULL && LAST_POS(f) == io_u->offset)
  89                 return 0;
  90
  91         if (lseek(f->fd, io_u->offset, SEEK_SET) == -1) {
  92                 td_verror(td, errno, "lseek");
  93                 return 1;
  94         }
  95
  96         return 0;
  97 }
  98
  99 static int fio_io_end(struct thread_data *td, struct io_u *io_u, int ret)
 100 {
 101         if (io_u->file && ret >= 0 && ddir_rw(io_u->ddir))
 102                 LAST_POS(io_u->file) = io_u->offset + ret;
 103
 104         if (ret != (int) io_u->xfer_buflen) {
 105                 if (ret >= 0) {
 106                         io_u->resid = io_u->xfer_buflen - ret;
 107                         io_u->error = 0;
 108                         return FIO_Q_COMPLETED;
 109                 } else
 110                         io_u->error = errno;
 111         }
 112
 113         if (io_u->error) {
 114                 io_u_log_error(td, io_u);
 115                 td_verror(td, io_u->error, "xfer");
 116         }
 117
 118         return FIO_Q_COMPLETED;
 119 }
 120
 121 #ifdef CONFIG_PWRITEV
 122 static int fio_pvsyncio_queue(struct thread_data *td, struct io_u *io_u)
 123 {
 124         struct syncio_data *sd = td->io_ops_data;
 125         struct iovec *iov = &sd->iovecs[0];
 126         struct fio_file *f = io_u->file;
 127         int ret;
 128
 129         fio_ro_check(td, io_u);
 130
 131         iov->iov_base = io_u->xfer_buf;
 132         iov->iov_len = io_u->xfer_buflen;
 133
 134         if (io_u->ddir == DDIR_READ)
 135                 ret = preadv(f->fd, iov, 1, io_u->offset);
 136         else if (io_u->ddir == DDIR_WRITE)
 137                 ret = pwritev(f->fd, iov, 1, io_u->offset);
 138         else if (io_u->ddir == DDIR_TRIM) {
 139                 do_io_u_trim(td, io_u);
 140                 return FIO_Q_COMPLETED;
 141         } else
 142                 ret = do_io_u_sync(td, io_u);
 143
 144         return fio_io_end(td, io_u, ret);
 145 }
 146 #endif
 147
 148 #ifdef FIO_HAVE_PWRITEV2
 149 static int fio_pvsyncio2_queue(struct thread_data *td, struct io_u *io_u)
 150 {
 151         struct syncio_data *sd = td->io_ops_data;
 152         struct psyncv2_options *o = td->eo;
 153         struct iovec *iov = &sd->iovecs[0];
 154         struct fio_file *f = io_u->file;
 155         int ret, flags = 0;
 156
 157         fio_ro_check(td, io_u);
 158
 159         if (o->hipri)
 160                 flags |= RWF_HIPRI;
 161         if (o->stream)
 162                 flags |= o->stream;
 163
 164         iov->iov_base = io_u->xfer_buf;
 165         iov->iov_len = io_u->xfer_buflen;
 166
 167         if (io_u->ddir == DDIR_READ)
 168                 ret = preadv2(f->fd, iov, 1, io_u->offset, flags);
 169         else if (io_u->ddir == DDIR_WRITE)
 170                 ret = pwritev2(f->fd, iov, 1, io_u->offset, flags);
 171         else if (io_u->ddir == DDIR_TRIM) {
 172                 do_io_u_trim(td, io_u);
 173                 return FIO_Q_COMPLETED;
 174         } else
 175                 ret = do_io_u_sync(td, io_u);
 176
 177         return fio_io_end(td, io_u, ret);
 178 }
 179 #endif
 180
 181
 182 static int fio_psyncio_queue(struct thread_data *td, struct io_u *io_u)
 183 {
 184         struct fio_file *f = io_u->file;
 185         int ret;
 186
 187         fio_ro_check(td, io_u);
 188
 189         if (io_u->ddir == DDIR_READ)
 190                 ret = pread(f->fd, io_u->xfer_buf, io_u->xfer_buflen, io_u->offset);
 191         else if (io_u->ddir == DDIR_WRITE)
 192                 ret = pwrite(f->fd, io_u->xfer_buf, io_u->xfer_buflen, io_u->offset);
 193         else if (io_u->ddir == DDIR_TRIM) {
 194                 do_io_u_trim(td, io_u);
 195                 return FIO_Q_COMPLETED;
 196         } else
 197                 ret = do_io_u_sync(td, io_u);
 198
 199         return fio_io_end(td, io_u, ret);
 200 }
 201
 202 static int fio_syncio_queue(struct thread_data *td, struct io_u *io_u)
 203 {
 204         struct fio_file *f = io_u->file;
 205         int ret;
 206
 207         fio_ro_check(td, io_u);
 208
 209         if (io_u->ddir == DDIR_READ)
 210                 ret = read(f->fd, io_u->xfer_buf, io_u->xfer_buflen);
 211         else if (io_u->ddir == DDIR_WRITE)
 212                 ret = write(f->fd, io_u->xfer_buf, io_u->xfer_buflen);
 213         else if (io_u->ddir == DDIR_TRIM) {
 214                 do_io_u_trim(td, io_u);
 215                 return FIO_Q_COMPLETED;
 216         } else
 217                 ret = do_io_u_sync(td, io_u);
 218
 219         return fio_io_end(td, io_u, ret);
 220 }
 221
 222 static int fio_vsyncio_getevents(struct thread_data *td, unsigned int min,
 223                                  unsigned int max,
 224                                  const struct timespec fio_unused *t)
 225 {
 226         struct syncio_data *sd = td->io_ops_data;
 227         int ret;
 228
 229         if (min) {
 230                 ret = sd->events;
 231                 sd->events = 0;
 232         } else
 233                 ret = 0;
 234
 235         dprint(FD_IO, "vsyncio_getevents: min=%d,max=%d: %d\n", min, max, ret);
 236         return ret;
 237 }
 238
 239 static struct io_u *fio_vsyncio_event(struct thread_data *td, int event)
 240 {
 241         struct syncio_data *sd = td->io_ops_data;
 242
 243         return sd->io_us[event];
 244 }
 245
 246 static int fio_vsyncio_append(struct thread_data *td, struct io_u *io_u)
 247 {
 248         struct syncio_data *sd = td->io_ops_data;
 249
 250         if (ddir_sync(io_u->ddir))
 251                 return 0;
 252
 253         if (io_u->offset == sd->last_offset && io_u->file == sd->last_file &&
 254             io_u->ddir == sd->last_ddir)
 255                 return 1;
 256
 257         return 0;
 258 }
 259
 260 static void fio_vsyncio_set_iov(struct syncio_data *sd, struct io_u *io_u,
 261                                 int idx)
 262 {
 263         sd->io_us[idx] = io_u;
 264         sd->iovecs[idx].iov_base = io_u->xfer_buf;
 265         sd->iovecs[idx].iov_len = io_u->xfer_buflen;
 266         sd->last_offset = io_u->offset + io_u->xfer_buflen;
 267         sd->last_file = io_u->file;
 268         sd->last_ddir = io_u->ddir;
 269         sd->queued_bytes += io_u->xfer_buflen;
 270         sd->queued++;
 271 }
 272
 273 static int fio_vsyncio_queue(struct thread_data *td, struct io_u *io_u)
 274 {
 275         struct syncio_data *sd = td->io_ops_data;
 276
 277         fio_ro_check(td, io_u);
 278
 279         if (!fio_vsyncio_append(td, io_u)) {
 280                 dprint(FD_IO, "vsyncio_queue: no append (%d)\n", sd->queued);
 281                 /*
 282                  * If we can't append and have stuff queued, tell fio to
 283                  * commit those first and then retry this io
 284                  */
 285                 if (sd->queued)
 286                         return FIO_Q_BUSY;
 287                 if (ddir_sync(io_u->ddir)) {
 288                         int ret = do_io_u_sync(td, io_u);
 289
 290                         return fio_io_end(td, io_u, ret);
 291                 }
 292
 293                 sd->queued = 0;
 294                 sd->queued_bytes = 0;
 295                 fio_vsyncio_set_iov(sd, io_u, 0);
 296         } else {
 297                 if (sd->queued == td->o.iodepth) {
 298                         dprint(FD_IO, "vsyncio_queue: max depth %d\n", sd->queued);
 299                         return FIO_Q_BUSY;
 300                 }
 301
 302                 dprint(FD_IO, "vsyncio_queue: append\n");
 303                 fio_vsyncio_set_iov(sd, io_u, sd->queued);
 304         }
 305
 306         dprint(FD_IO, "vsyncio_queue: depth now %d\n", sd->queued);
 307         return FIO_Q_QUEUED;
 308 }
 309
 310 /*
 311  * Check that we transferred all bytes, or saw an error, etc
 312  */
 313 static int fio_vsyncio_end(struct thread_data *td, ssize_t bytes)
 314 {
 315         struct syncio_data *sd = td->io_ops_data;
 316         struct io_u *io_u;
 317         unsigned int i;
 318         int err;
 319
 320         /*
 321          * transferred everything, perfect
 322          */
 323         if (bytes == sd->queued_bytes)
 324                 return 0;
 325
 326         err = errno;
 327         for (i = 0; i < sd->queued; i++) {
 328                 io_u = sd->io_us[i];
 329
 330                 if (bytes == -1) {
 331                         io_u->error = err;
 332                 } else {
 333                         unsigned int this_io;
 334
 335                         this_io = bytes;
 336                         if (this_io > io_u->xfer_buflen)
 337                                 this_io = io_u->xfer_buflen;
 338
 339                         io_u->resid = io_u->xfer_buflen - this_io;
 340                         io_u->error = 0;
 341                         bytes -= this_io;
 342                 }
 343         }
 344
 345         if (bytes == -1) {
 346                 td_verror(td, err, "xfer vsync");
 347                 return -err;
 348         }
 349
 350         return 0;
 351 }
 352
 353 static int fio_vsyncio_commit(struct thread_data *td)
 354 {
 355         struct syncio_data *sd = td->io_ops_data;
 356         struct fio_file *f;
 357         ssize_t ret;
 358
 359         if (!sd->queued)
 360                 return 0;
 361
 362         io_u_mark_submit(td, sd->queued);
 363         f = sd->last_file;
 364
 365         if (lseek(f->fd, sd->io_us[0]->offset, SEEK_SET) == -1) {
 366                 int err = -errno;
 367
 368                 td_verror(td, errno, "lseek");
 369                 return err;
 370         }
 371
 372         if (sd->last_ddir == DDIR_READ)
 373                 ret = readv(f->fd, sd->iovecs, sd->queued);
 374         else
 375                 ret = writev(f->fd, sd->iovecs, sd->queued);
 376
 377         dprint(FD_IO, "vsyncio_commit: %d\n", (int) ret);
 378         sd->events = sd->queued;
 379         sd->queued = 0;
 380         return fio_vsyncio_end(td, ret);
 381 }
 382
 383 static int fio_vsyncio_init(struct thread_data *td)
 384 {
 385         struct syncio_data *sd;
 386
 387         sd = malloc(sizeof(*sd));
 388         memset(sd, 0, sizeof(*sd));
 389         sd->last_offset = -1ULL;
 390         sd->iovecs = malloc(td->o.iodepth * sizeof(struct iovec));
 391         sd->io_us = malloc(td->o.iodepth * sizeof(struct io_u *));
 392
 393         td->io_ops_data = sd;
 394         return 0;
 395 }
 396
 397 static void fio_vsyncio_cleanup(struct thread_data *td)
 398 {
 399         struct syncio_data *sd = td->io_ops_data;
 400
 401         if (sd) {
 402                 free(sd->iovecs);
 403                 free(sd->io_us);
 404                 free(sd);
 405         }
 406 }
 407
 408 static struct ioengine_ops ioengine_rw = {
 409         .name           = "sync",
 410         .version        = FIO_IOOPS_VERSION,
 411         .prep           = fio_syncio_prep,
 412         .queue          = fio_syncio_queue,
 413         .open_file      = generic_open_file,
 414         .close_file     = generic_close_file,
 415         .get_file_size  = generic_get_file_size,
 416         .flags          = FIO_SYNCIO,
 417 };
 418
 419 static struct ioengine_ops ioengine_prw = {
 420         .name           = "psync",
 421         .version        = FIO_IOOPS_VERSION,
 422         .queue          = fio_psyncio_queue,
 423         .open_file      = generic_open_file,
 424         .close_file     = generic_close_file,
 425         .get_file_size  = generic_get_file_size,
 426         .flags          = FIO_SYNCIO,
 427 };
 428
 429 static struct ioengine_ops ioengine_vrw = {
 430         .name           = "vsync",
 431         .version        = FIO_IOOPS_VERSION,
 432         .init           = fio_vsyncio_init,
 433         .cleanup        = fio_vsyncio_cleanup,
 434         .queue          = fio_vsyncio_queue,
 435         .commit         = fio_vsyncio_commit,
 436         .event          = fio_vsyncio_event,
 437         .getevents      = fio_vsyncio_getevents,
 438         .open_file      = generic_open_file,
 439         .close_file     = generic_close_file,
 440         .get_file_size  = generic_get_file_size,
 441         .flags          = FIO_SYNCIO,
 442 };
 443
 444 #ifdef CONFIG_PWRITEV
 445 static struct ioengine_ops ioengine_pvrw = {
 446         .name           = "pvsync",
 447         .version        = FIO_IOOPS_VERSION,
 448         .init           = fio_vsyncio_init,
 449         .cleanup        = fio_vsyncio_cleanup,
 450         .queue          = fio_pvsyncio_queue,
 451         .open_file      = generic_open_file,
 452         .close_file     = generic_close_file,
 453         .get_file_size  = generic_get_file_size,
 454         .flags          = FIO_SYNCIO,
 455 };
 456 #endif
 457
 458 #ifdef FIO_HAVE_PWRITEV2
 459 static struct ioengine_ops ioengine_pvrw2 = {
 460         .name           = "pvsync2",
 461         .version        = FIO_IOOPS_VERSION,
 462         .init           = fio_vsyncio_init,
 463         .cleanup        = fio_vsyncio_cleanup,
 464         .queue          = fio_pvsyncio2_queue,
 465         .open_file      = generic_open_file,
 466         .close_file     = generic_close_file,
 467         .get_file_size  = generic_get_file_size,
 468         .flags          = FIO_SYNCIO,
 469         .options        = options,
 470         .option_struct_size     = sizeof(struct psyncv2_options),
 471 };
 472 #endif
 473
 474 static void fio_init fio_syncio_register(void)
 475 {
 476         register_ioengine(&ioengine_rw);
 477         register_ioengine(&ioengine_prw);
 478         register_ioengine(&ioengine_vrw);
 479 #ifdef CONFIG_PWRITEV
 480         register_ioengine(&ioengine_pvrw);
 481 #endif
 482 #ifdef FIO_HAVE_PWRITEV2
 483         register_ioengine(&ioengine_pvrw2);
 484 #endif
 485 }
 486
 487 static void fio_exit fio_syncio_unregister(void)
 488 {
 489         unregister_ioengine(&ioengine_rw);
 490         unregister_ioengine(&ioengine_prw);
 491         unregister_ioengine(&ioengine_vrw);
 492 #ifdef CONFIG_PWRITEV
 493         unregister_ioengine(&ioengine_pvrw);
 494 #endif
 495 #ifdef FIO_HAVE_PWRITEV2
 496         unregister_ioengine(&ioengine_pvrw2);
 497 #endif
 498 }