engines/rbd.c

   1 /*
   2  * rbd engine
   3  *
   4  * IO engine using Ceph's librbd to test RADOS Block Devices.
   5  *
   6  */
   7
   8 #include <rbd/librbd.h>
   9
  10 #include "../fio.h"
  11
  12 struct fio_rbd_iou {
  13         struct io_u *io_u;
  14         rbd_completion_t completion;
  15         int io_complete;
  16         int io_seen;
  17 };
  18
  19 struct rbd_data {
  20         rados_t cluster;
  21         rados_ioctx_t io_ctx;
  22         rbd_image_t image;
  23         struct io_u **aio_events;
  24 };
  25
  26 struct rbd_options {
  27         struct thread_data *td;
  28         char *rbd_name;
  29         char *pool_name;
  30         char *client_name;
  31 };
  32
  33 static struct fio_option options[] = {
  34         {
  35                 .name           = "rbdname",
  36                 .lname          = "rbd engine rbdname",
  37                 .type           = FIO_OPT_STR_STORE,
  38                 .help           = "RBD name for RBD engine",
  39                 .off1           = offsetof(struct rbd_options, rbd_name),
  40                 .category       = FIO_OPT_C_ENGINE,
  41                 .group          = FIO_OPT_G_RBD,
  42         },
  43         {
  44                 .name     = "pool",
  45                 .lname    = "rbd engine pool",
  46                 .type     = FIO_OPT_STR_STORE,
  47                 .help     = "Name of the pool hosting the RBD for the RBD engine",
  48                 .off1     = offsetof(struct rbd_options, pool_name),
  49                 .category = FIO_OPT_C_ENGINE,
  50                 .group    = FIO_OPT_G_RBD,
  51         },
  52         {
  53                 .name     = "clientname",
  54                 .lname    = "rbd engine clientname",
  55                 .type     = FIO_OPT_STR_STORE,
  56                 .help     = "Name of the ceph client to access the RBD for the RBD engine",
  57                 .off1     = offsetof(struct rbd_options, client_name),
  58                 .category = FIO_OPT_C_ENGINE,
  59                 .group    = FIO_OPT_G_RBD,
  60         },
  61         {
  62                 .name = NULL,
  63         },
  64 };
  65
  66 static int _fio_setup_rbd_data(struct thread_data *td,
  67                                struct rbd_data **rbd_data_ptr)
  68 {
  69         struct rbd_data *rbd_data;
  70
  71         if (td->io_ops->data)
  72                 return 0;
  73
  74         rbd_data = malloc(sizeof(struct rbd_data));
  75         if (!rbd_data)
  76                 goto failed;
  77
  78         memset(rbd_data, 0, sizeof(struct rbd_data));
  79
  80         rbd_data->aio_events = malloc(td->o.iodepth * sizeof(struct io_u *));
  81         if (!rbd_data->aio_events)
  82                 goto failed;
  83
  84         memset(rbd_data->aio_events, 0, td->o.iodepth * sizeof(struct io_u *));
  85
  86         *rbd_data_ptr = rbd_data;
  87
  88         return 0;
  89
  90 failed:
  91         if (rbd_data)
  92                 free(rbd_data);
  93         return 1;
  94
  95 }
  96
  97 static int _fio_rbd_connect(struct thread_data *td)
  98 {
  99         struct rbd_data *rbd_data = td->io_ops->data;
 100         struct rbd_options *o = td->eo;
 101         int r;
 102
 103         r = rados_create(&(rbd_data->cluster), o->client_name);
 104         if (r < 0) {
 105                 log_err("rados_create failed.\n");
 106                 goto failed_early;
 107         }
 108
 109         r = rados_conf_read_file(rbd_data->cluster, NULL);
 110         if (r < 0) {
 111                 log_err("rados_conf_read_file failed.\n");
 112                 goto failed_early;
 113         }
 114
 115         r = rados_connect(rbd_data->cluster);
 116         if (r < 0) {
 117                 log_err("rados_connect failed.\n");
 118                 goto failed_shutdown;
 119         }
 120
 121         r = rados_ioctx_create(rbd_data->cluster, o->pool_name,
 122                                &(rbd_data->io_ctx));
 123         if (r < 0) {
 124                 log_err("rados_ioctx_create failed.\n");
 125                 goto failed_shutdown;
 126         }
 127
 128         r = rbd_open(rbd_data->io_ctx, o->rbd_name, &(rbd_data->image),
 129                      NULL /*snap */ );
 130         if (r < 0) {
 131                 log_err("rbd_open failed.\n");
 132                 goto failed_open;
 133         }
 134         return 0;
 135
 136 failed_open:
 137         rados_ioctx_destroy(rbd_data->io_ctx);
 138         rbd_data->io_ctx = NULL;
 139 failed_shutdown:
 140         rados_shutdown(rbd_data->cluster);
 141         rbd_data->cluster = NULL;
 142 failed_early:
 143         return 1;
 144 }
 145
 146 static void _fio_rbd_disconnect(struct rbd_data *rbd_data)
 147 {
 148         if (!rbd_data)
 149                 return;
 150
 151         /* shutdown everything */
 152         if (rbd_data->image) {
 153                 rbd_close(rbd_data->image);
 154                 rbd_data->image = NULL;
 155         }
 156
 157         if (rbd_data->io_ctx) {
 158                 rados_ioctx_destroy(rbd_data->io_ctx);
 159                 rbd_data->io_ctx = NULL;
 160         }
 161
 162         if (rbd_data->cluster) {
 163                 rados_shutdown(rbd_data->cluster);
 164                 rbd_data->cluster = NULL;
 165         }
 166 }
 167
 168 static void _fio_rbd_finish_aiocb(rbd_completion_t comp, void *data)
 169 {
 170         struct io_u *io_u = data;
 171         struct fio_rbd_iou *fri = io_u->engine_data;
 172         ssize_t ret;
 173
 174         fri->io_complete = 1;
 175
 176         /*
 177          * Looks like return value is 0 for success, or < 0 for
 178          * a specific error. So we have to assume that it can't do
 179          * partial completions.
 180          */
 181         ret = rbd_aio_get_return_value(fri->completion);
 182         if (ret < 0) {
 183                 io_u->error = ret;
 184                 io_u->resid = io_u->xfer_buflen;
 185         } else
 186                 io_u->error = 0;
 187 }
 188
 189 static struct io_u *fio_rbd_event(struct thread_data *td, int event)
 190 {
 191         struct rbd_data *rbd_data = td->io_ops->data;
 192
 193         return rbd_data->aio_events[event];
 194 }
 195
 196 static inline int fri_check_complete(struct rbd_data *rbd_data,
 197                                      struct io_u *io_u,
 198                                      unsigned int *events)
 199 {
 200         struct fio_rbd_iou *fri = io_u->engine_data;
 201
 202         if (fri->io_complete) {
 203                 fri->io_complete = 0;
 204                 fri->io_seen = 1;
 205                 rbd_data->aio_events[*events] = io_u;
 206                 (*events)++;
 207
 208                 rbd_aio_release(fri->completion);
 209                 return 1;
 210         }
 211
 212         return 0;
 213 }
 214
 215 static int rbd_iter_events(struct thread_data *td, unsigned int *events,
 216                            unsigned int min_evts, int wait)
 217 {
 218         struct rbd_data *rbd_data = td->io_ops->data;
 219         unsigned int this_events = 0;
 220         struct io_u *io_u;
 221         int i;
 222
 223         io_u_qiter(&td->io_u_all, io_u, i) {
 224                 struct fio_rbd_iou *fri = io_u->engine_data;
 225
 226                 if (!(io_u->flags & IO_U_F_FLIGHT))
 227                         continue;
 228                 if (fri->io_seen)
 229                         continue;
 230
 231                 if (fri_check_complete(rbd_data, io_u, events))
 232                         this_events++;
 233                 else if (wait) {
 234                         rbd_aio_wait_for_complete(fri->completion);
 235
 236                         if (fri_check_complete(rbd_data, io_u, events))
 237                                 this_events++;
 238                 }
 239                 if (*events >= min_evts)
 240                         break;
 241         }
 242
 243         return this_events;
 244 }
 245
 246 static int fio_rbd_getevents(struct thread_data *td, unsigned int min,
 247                              unsigned int max, const struct timespec *t)
 248 {
 249         unsigned int this_events, events = 0;
 250         int wait = 0;
 251
 252         do {
 253                 this_events = rbd_iter_events(td, &events, min, wait);
 254
 255                 if (events >= min)
 256                         break;
 257                 if (this_events)
 258                         continue;
 259
 260                 wait = 1;
 261         } while (1);
 262
 263         return events;
 264 }
 265
 266 static int fio_rbd_queue(struct thread_data *td, struct io_u *io_u)
 267 {
 268         struct rbd_data *rbd_data = td->io_ops->data;
 269         struct fio_rbd_iou *fri = io_u->engine_data;
 270         int r = -1;
 271
 272         fio_ro_check(td, io_u);
 273
 274         fri->io_complete = 0;
 275         fri->io_seen = 0;
 276
 277         if (io_u->ddir == DDIR_WRITE) {
 278                 r = rbd_aio_create_completion(io_u, _fio_rbd_finish_aiocb,
 279                                                 &fri->completion);
 280                 if (r < 0) {
 281                         log_err
 282                             ("rbd_aio_create_completion for DDIR_WRITE failed.\n");
 283                         goto failed;
 284                 }
 285
 286                 r = rbd_aio_write(rbd_data->image, io_u->offset,
 287                                   io_u->xfer_buflen, io_u->xfer_buf,
 288                                   fri->completion);
 289                 if (r < 0) {
 290                         log_err("rbd_aio_write failed.\n");
 291                         rbd_aio_release(fri->completion);
 292                         goto failed;
 293                 }
 294
 295         } else if (io_u->ddir == DDIR_READ) {
 296                 r = rbd_aio_create_completion(io_u, _fio_rbd_finish_aiocb,
 297                                                 &fri->completion);
 298                 if (r < 0) {
 299                         log_err
 300                             ("rbd_aio_create_completion for DDIR_READ failed.\n");
 301                         goto failed;
 302                 }
 303
 304                 r = rbd_aio_read(rbd_data->image, io_u->offset,
 305                                  io_u->xfer_buflen, io_u->xfer_buf,
 306                                  fri->completion);
 307
 308                 if (r < 0) {
 309                         log_err("rbd_aio_read failed.\n");
 310                         rbd_aio_release(fri->completion);
 311                         goto failed;
 312                 }
 313
 314         } else if (io_u->ddir == DDIR_SYNC) {
 315                 r = rbd_aio_create_completion(io_u, _fio_rbd_finish_aiocb,
 316                                                 &fri->completion);
 317                 if (r < 0) {
 318                         log_err
 319                             ("rbd_aio_create_completion for DDIR_SYNC failed.\n");
 320                         goto failed;
 321                 }
 322
 323                 r = rbd_aio_flush(rbd_data->image, fri->completion);
 324                 if (r < 0) {
 325                         log_err("rbd_flush failed.\n");
 326                         rbd_aio_release(fri->completion);
 327                         goto failed;
 328                 }
 329
 330         } else {
 331                 dprint(FD_IO, "%s: Warning: unhandled ddir: %d\n", __func__,
 332                        io_u->ddir);
 333                 return FIO_Q_COMPLETED;
 334         }
 335
 336         return FIO_Q_QUEUED;
 337
 338 failed:
 339         io_u->error = r;
 340         td_verror(td, io_u->error, "xfer");
 341         return FIO_Q_COMPLETED;
 342 }
 343
 344 static int fio_rbd_init(struct thread_data *td)
 345 {
 346         int r;
 347
 348         r = _fio_rbd_connect(td);
 349         if (r) {
 350                 log_err("fio_rbd_connect failed, return code: %d .\n", r);
 351                 goto failed;
 352         }
 353
 354         return 0;
 355
 356 failed:
 357         return 1;
 358 }
 359
 360 static void fio_rbd_cleanup(struct thread_data *td)
 361 {
 362         struct rbd_data *rbd_data = td->io_ops->data;
 363
 364         if (rbd_data) {
 365                 _fio_rbd_disconnect(rbd_data);
 366                 free(rbd_data->aio_events);
 367                 free(rbd_data);
 368         }
 369
 370 }
 371
 372 static int fio_rbd_setup(struct thread_data *td)
 373 {
 374         int r = 0;
 375         rbd_image_info_t info;
 376         struct fio_file *f;
 377         struct rbd_data *rbd_data = NULL;
 378         int major, minor, extra;
 379
 380         /* log version of librbd. No cluster connection required. */
 381         rbd_version(&major, &minor, &extra);
 382         log_info("rbd engine: RBD version: %d.%d.%d\n", major, minor, extra);
 383
 384         /* allocate engine specific structure to deal with librbd. */
 385         r = _fio_setup_rbd_data(td, &rbd_data);
 386         if (r) {
 387                 log_err("fio_setup_rbd_data failed.\n");
 388                 goto cleanup;
 389         }
 390         td->io_ops->data = rbd_data;
 391
 392         /* librbd does not allow us to run first in the main thread and later
 393          * in a fork child. It needs to be the same process context all the
 394          * time.
 395          */
 396         td->o.use_thread = 1;
 397
 398         /* connect in the main thread to determine to determine
 399          * the size of the given RADOS block device. And disconnect
 400          * later on.
 401          */
 402         r = _fio_rbd_connect(td);
 403         if (r) {
 404                 log_err("fio_rbd_connect failed.\n");
 405                 goto cleanup;
 406         }
 407
 408         /* get size of the RADOS block device */
 409         r = rbd_stat(rbd_data->image, &info, sizeof(info));
 410         if (r < 0) {
 411                 log_err("rbd_status failed.\n");
 412                 goto disconnect;
 413         }
 414         dprint(FD_IO, "rbd-engine: image size: %lu\n", info.size);
 415
 416         /* taken from "net" engine. Pretend we deal with files,
 417          * even if we do not have any ideas about files.
 418          * The size of the RBD is set instead of a artificial file.
 419          */
 420         if (!td->files_index) {
 421                 add_file(td, td->o.filename ? : "rbd", 0, 0);
 422                 td->o.nr_files = td->o.nr_files ? : 1;
 423                 td->o.open_files++;
 424         }
 425         f = td->files[0];
 426         f->real_file_size = info.size;
 427
 428         /* disconnect, then we were only connected to determine
 429          * the size of the RBD.
 430          */
 431         _fio_rbd_disconnect(rbd_data);
 432         return 0;
 433
 434 disconnect:
 435         _fio_rbd_disconnect(rbd_data);
 436 cleanup:
 437         fio_rbd_cleanup(td);
 438         return r;
 439 }
 440
 441 static int fio_rbd_open(struct thread_data *td, struct fio_file *f)
 442 {
 443         return 0;
 444 }
 445
 446 static int fio_rbd_invalidate(struct thread_data *td, struct fio_file *f)
 447 {
 448 #if defined(CONFIG_RBD_INVAL)
 449         struct rbd_data *rbd_data = td->io_ops->data;
 450
 451         return rbd_invalidate_cache(rbd_data->image);
 452 #else
 453         return 0;
 454 #endif
 455 }
 456
 457 static void fio_rbd_io_u_free(struct thread_data *td, struct io_u *io_u)
 458 {
 459         struct fio_rbd_iou *fri = io_u->engine_data;
 460
 461         if (fri) {
 462                 io_u->engine_data = NULL;
 463                 free(fri);
 464         }
 465 }
 466
 467 static int fio_rbd_io_u_init(struct thread_data *td, struct io_u *io_u)
 468 {
 469         struct fio_rbd_iou *fri;
 470
 471         fri = calloc(1, sizeof(*fri));
 472         fri->io_u = io_u;
 473         io_u->engine_data = fri;
 474         return 0;
 475 }
 476
 477 static struct ioengine_ops ioengine = {
 478         .name                   = "rbd",
 479         .version                = FIO_IOOPS_VERSION,
 480         .setup                  = fio_rbd_setup,
 481         .init                   = fio_rbd_init,
 482         .queue                  = fio_rbd_queue,
 483         .getevents              = fio_rbd_getevents,
 484         .event                  = fio_rbd_event,
 485         .cleanup                = fio_rbd_cleanup,
 486         .open_file              = fio_rbd_open,
 487         .invalidate             = fio_rbd_invalidate,
 488         .options                = options,
 489         .io_u_init              = fio_rbd_io_u_init,
 490         .io_u_free              = fio_rbd_io_u_free,
 491         .option_struct_size     = sizeof(struct rbd_options),
 492 };
 493
 494 static void fio_init fio_rbd_register(void)
 495 {
 496         register_ioengine(&ioengine);
 497 }
 498
 499 static void fio_exit fio_rbd_unregister(void)
 500 {
 501         unregister_ioengine(&ioengine);
 502 }