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 failed_shutdown:
 135         rados_shutdown(rbd_data->cluster);
 136 failed_early:
 137         return 1;
 138 }
 139
 140 static void _fio_rbd_disconnect(struct rbd_data *rbd_data)
 141 {
 142         if (!rbd_data)
 143                 return;
 144
 145         /* shutdown everything */
 146         if (rbd_data->image) {
 147                 rbd_close(rbd_data->image);
 148                 rbd_data->image = NULL;
 149         }
 150
 151         if (rbd_data->io_ctx) {
 152                 rados_ioctx_destroy(rbd_data->io_ctx);
 153                 rbd_data->io_ctx = NULL;
 154         }
 155
 156         if (rbd_data->cluster) {
 157                 rados_shutdown(rbd_data->cluster);
 158                 rbd_data->cluster = NULL;
 159         }
 160 }
 161
 162 static void _fio_rbd_finish_write_aiocb(rbd_completion_t comp, void *data)
 163 {
 164         struct io_u *io_u = (struct io_u *)data;
 165         struct fio_rbd_iou *fio_rbd_iou =
 166             (struct fio_rbd_iou *)io_u->engine_data;
 167
 168         fio_rbd_iou->io_complete = 1;
 169
 170         /* if write needs to be verified - we should not release comp here
 171            without fetching the result */
 172
 173         rbd_aio_release(comp);
 174         /* TODO handle error */
 175
 176         return;
 177 }
 178
 179 static void _fio_rbd_finish_read_aiocb(rbd_completion_t comp, void *data)
 180 {
 181         struct io_u *io_u = (struct io_u *)data;
 182         struct fio_rbd_iou *fio_rbd_iou =
 183             (struct fio_rbd_iou *)io_u->engine_data;
 184
 185         fio_rbd_iou->io_complete = 1;
 186
 187         /* if read needs to be verified - we should not release comp here
 188            without fetching the result */
 189         rbd_aio_release(comp);
 190
 191         /* TODO handle error */
 192
 193         return;
 194 }
 195
 196 static struct io_u *fio_rbd_event(struct thread_data *td, int event)
 197 {
 198         struct rbd_data *rbd_data = td->io_ops->data;
 199
 200         return rbd_data->aio_events[event];
 201 }
 202
 203 static int fio_rbd_getevents(struct thread_data *td, unsigned int min,
 204                              unsigned int max, struct timespec *t)
 205 {
 206         struct rbd_data *rbd_data = td->io_ops->data;
 207         unsigned int events = 0;
 208         struct io_u *io_u;
 209         int i;
 210         struct fio_rbd_iou *fov;
 211
 212         do {
 213                 io_u_qiter(&td->io_u_all, io_u, i) {
 214                         if (!(io_u->flags & IO_U_F_FLIGHT))
 215                                 continue;
 216
 217                         fov = (struct fio_rbd_iou *)io_u->engine_data;
 218
 219                         if (fov->io_complete) {
 220                                 fov->io_complete = 0;
 221                                 rbd_data->aio_events[events] = io_u;
 222                                 events++;
 223                         }
 224
 225                 }
 226                 if (events < min)
 227                         usleep(100);
 228                 else
 229                         break;
 230
 231         } while (1);
 232
 233         return events;
 234 }
 235
 236 static int fio_rbd_queue(struct thread_data *td, struct io_u *io_u)
 237 {
 238         int r = -1;
 239         struct rbd_data *rbd_data = td->io_ops->data;
 240         rbd_completion_t comp;
 241
 242         fio_ro_check(td, io_u);
 243
 244         if (io_u->ddir == DDIR_WRITE) {
 245                 r = rbd_aio_create_completion(io_u,
 246                                               (rbd_callback_t)
 247                                               _fio_rbd_finish_write_aiocb,
 248                                               &comp);
 249                 if (r < 0) {
 250                         log_err
 251                             ("rbd_aio_create_completion for DDIR_WRITE failed.\n");
 252                         goto failed;
 253                 }
 254
 255                 r = rbd_aio_write(rbd_data->image, io_u->offset,
 256                                   io_u->xfer_buflen, io_u->xfer_buf, comp);
 257                 if (r < 0) {
 258                         log_err("rbd_aio_write failed.\n");
 259                         goto failed;
 260                 }
 261
 262         } else if (io_u->ddir == DDIR_READ) {
 263                 r = rbd_aio_create_completion(io_u,
 264                                               (rbd_callback_t)
 265                                               _fio_rbd_finish_read_aiocb,
 266                                               &comp);
 267                 if (r < 0) {
 268                         log_err
 269                             ("rbd_aio_create_completion for DDIR_READ failed.\n");
 270                         goto failed;
 271                 }
 272
 273                 r = rbd_aio_read(rbd_data->image, io_u->offset,
 274                                  io_u->xfer_buflen, io_u->xfer_buf, comp);
 275
 276                 if (r < 0) {
 277                         log_err("rbd_aio_read failed.\n");
 278                         goto failed;
 279                 }
 280
 281         } else if (io_u->ddir == DDIR_SYNC) {
 282                 r = rbd_flush(rbd_data->image);
 283                 if (r < 0) {
 284                         log_err("rbd_flush failed.\n");
 285                         goto failed;
 286                 }
 287
 288                 return FIO_Q_COMPLETED;
 289         } else {
 290                 dprint(FD_IO, "%s: Warning: unhandled ddir: %d\n", __func__,
 291                        io_u->ddir);
 292                 return FIO_Q_COMPLETED;
 293         }
 294
 295         return FIO_Q_QUEUED;
 296
 297 failed:
 298         io_u->error = r;
 299         td_verror(td, io_u->error, "xfer");
 300         return FIO_Q_COMPLETED;
 301 }
 302
 303 static int fio_rbd_init(struct thread_data *td)
 304 {
 305         int r;
 306
 307         r = _fio_rbd_connect(td);
 308         if (r) {
 309                 log_err("fio_rbd_connect failed, return code: %d .\n", r);
 310                 goto failed;
 311         }
 312
 313         return 0;
 314
 315 failed:
 316         return 1;
 317
 318 }
 319
 320 static void fio_rbd_cleanup(struct thread_data *td)
 321 {
 322         struct rbd_data *rbd_data = td->io_ops->data;
 323
 324         if (rbd_data) {
 325                 _fio_rbd_disconnect(rbd_data);
 326                 free(rbd_data->aio_events);
 327                 free(rbd_data);
 328         }
 329
 330 }
 331
 332 static int fio_rbd_setup(struct thread_data *td)
 333 {
 334         int r = 0;
 335         rbd_image_info_t info;
 336         struct fio_file *f;
 337         struct rbd_data *rbd_data = NULL;
 338         int major, minor, extra;
 339
 340         /* log version of librbd. No cluster connection required. */
 341         rbd_version(&major, &minor, &extra);
 342         log_info("rbd engine: RBD version: %d.%d.%d\n", major, minor, extra);
 343
 344         /* allocate engine specific structure to deal with librbd. */
 345         r = _fio_setup_rbd_data(td, &rbd_data);
 346         if (r) {
 347                 log_err("fio_setup_rbd_data failed.\n");
 348                 goto cleanup;
 349         }
 350         td->io_ops->data = rbd_data;
 351
 352         /* librbd does not allow us to run first in the main thread and later in a
 353          * fork child. It needs to be the same process context all the time.
 354          */
 355         td->o.use_thread = 1;
 356
 357         /* connect in the main thread to determine to determine
 358          * the size of the given RADOS block device. And disconnect
 359          * later on.
 360          */
 361         r = _fio_rbd_connect(td);
 362         if (r) {
 363                 log_err("fio_rbd_connect failed.\n");
 364                 goto cleanup;
 365         }
 366
 367         /* get size of the RADOS block device */
 368         r = rbd_stat(rbd_data->image, &info, sizeof(info));
 369         if (r < 0) {
 370                 log_err("rbd_status failed.\n");
 371                 goto disconnect;
 372         }
 373         dprint(FD_IO, "rbd-engine: image size: %lu\n", info.size);
 374
 375         /* taken from "net" engine. Pretend we deal with files,
 376          * even if we do not have any ideas about files.
 377          * The size of the RBD is set instead of a artificial file.
 378          */
 379         if (!td->files_index) {
 380                 add_file(td, td->o.filename ? : "rbd", 0, 0);
 381                 td->o.nr_files = td->o.nr_files ? : 1;
 382                 td->o.open_files++;
 383         }
 384         f = td->files[0];
 385         f->real_file_size = info.size;
 386
 387         /* disconnect, then we were only connected to determine
 388          * the size of the RBD.
 389          */
 390         _fio_rbd_disconnect(rbd_data);
 391         return 0;
 392
 393 disconnect:
 394         _fio_rbd_disconnect(rbd_data);
 395 cleanup:
 396         fio_rbd_cleanup(td);
 397         return r;
 398 }
 399
 400 static int fio_rbd_open(struct thread_data *td, struct fio_file *f)
 401 {
 402         return 0;
 403 }
 404
 405 static void fio_rbd_io_u_free(struct thread_data *td, struct io_u *io_u)
 406 {
 407         struct fio_rbd_iou *o = io_u->engine_data;
 408
 409         if (o) {
 410                 io_u->engine_data = NULL;
 411                 free(o);
 412         }
 413 }
 414
 415 static int fio_rbd_io_u_init(struct thread_data *td, struct io_u *io_u)
 416 {
 417         struct fio_rbd_iou *o;
 418
 419         o = malloc(sizeof(*o));
 420         o->io_complete = 0;
 421         o->io_u = io_u;
 422         io_u->engine_data = o;
 423         return 0;
 424 }
 425
 426 static struct ioengine_ops ioengine = {
 427         .name               = "rbd",
 428         .version            = FIO_IOOPS_VERSION,
 429         .setup              = fio_rbd_setup,
 430         .init               = fio_rbd_init,
 431         .queue              = fio_rbd_queue,
 432         .getevents          = fio_rbd_getevents,
 433         .event              = fio_rbd_event,
 434         .cleanup            = fio_rbd_cleanup,
 435         .open_file          = fio_rbd_open,
 436         .options            = options,
 437         .io_u_init          = fio_rbd_io_u_init,
 438         .io_u_free          = fio_rbd_io_u_free,
 439         .option_struct_size = sizeof(struct rbd_options),
 440 };
 441
 442 static void fio_init fio_rbd_register(void)
 443 {
 444         register_ioengine(&ioengine);
 445 }
 446
 447 static void fio_exit fio_rbd_unregister(void)
 448 {
 449         unregister_ioengine(&ioengine);
 450 }