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