block/rbd.c

   1 /*
   2  * QEMU Block driver for RADOS (Ceph)
   3  *
   4  * Copyright (C) 2010-2011 Christian Brunner <chb@muc.de>,
   5  *                         Josh Durgin <josh.durgin@dreamhost.com>
   6  *
   7  * This work is licensed under the terms of the GNU GPL, version 2.  See
   8  * the COPYING file in the top-level directory.
   9  *
  10  * Contributions after 2012-01-13 are licensed under the terms of the
  11  * GNU GPL, version 2 or (at your option) any later version.
  12  */
  13
  14 #include "qemu/osdep.h"
  15
  16 #include "qapi/error.h"
  17 #include "qemu/error-report.h"
  18 #include "block/block_int.h"
  19 #include "crypto/secret.h"
  20 #include "qemu/cutils.h"
  21
  22 #include <rbd/librbd.h>
  23
  24 /*
  25  * When specifying the image filename use:
  26  *
  27  * rbd:poolname/devicename[@snapshotname][:option1=value1[:option2=value2...]]
  28  *
  29  * poolname must be the name of an existing rados pool.
  30  *
  31  * devicename is the name of the rbd image.
  32  *
  33  * Each option given is used to configure rados, and may be any valid
  34  * Ceph option, "id", or "conf".
  35  *
  36  * The "id" option indicates what user we should authenticate as to
  37  * the Ceph cluster.  If it is excluded we will use the Ceph default
  38  * (normally 'admin').
  39  *
  40  * The "conf" option specifies a Ceph configuration file to read.  If
  41  * it is not specified, we will read from the default Ceph locations
  42  * (e.g., /etc/ceph/ceph.conf).  To avoid reading _any_ configuration
  43  * file, specify conf=/dev/null.
  44  *
  45  * Configuration values containing :, @, or = can be escaped with a
  46  * leading "\".
  47  */
  48
  49 /* rbd_aio_discard added in 0.1.2 */
  50 #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 2)
  51 #define LIBRBD_SUPPORTS_DISCARD
  52 #else
  53 #undef LIBRBD_SUPPORTS_DISCARD
  54 #endif
  55
  56 #define OBJ_MAX_SIZE (1UL << OBJ_DEFAULT_OBJ_ORDER)
  57
  58 #define RBD_MAX_CONF_NAME_SIZE 128
  59 #define RBD_MAX_CONF_VAL_SIZE 512
  60 #define RBD_MAX_CONF_SIZE 1024
  61 #define RBD_MAX_POOL_NAME_SIZE 128
  62 #define RBD_MAX_SNAP_NAME_SIZE 128
  63 #define RBD_MAX_SNAPS 100
  64
  65 /* The LIBRBD_SUPPORTS_IOVEC is defined in librbd.h */
  66 #ifdef LIBRBD_SUPPORTS_IOVEC
  67 #define LIBRBD_USE_IOVEC 1
  68 #else
  69 #define LIBRBD_USE_IOVEC 0
  70 #endif
  71
  72 typedef enum {
  73     RBD_AIO_READ,
  74     RBD_AIO_WRITE,
  75     RBD_AIO_DISCARD,
  76     RBD_AIO_FLUSH
  77 } RBDAIOCmd;
  78
  79 typedef struct RBDAIOCB {
  80     BlockAIOCB common;
  81     int64_t ret;
  82     QEMUIOVector *qiov;
  83     char *bounce;
  84     RBDAIOCmd cmd;
  85     int error;
  86     struct BDRVRBDState *s;
  87 } RBDAIOCB;
  88
  89 typedef struct RADOSCB {
  90     RBDAIOCB *acb;
  91     struct BDRVRBDState *s;
  92     int64_t size;
  93     char *buf;
  94     int64_t ret;
  95 } RADOSCB;
  96
  97 typedef struct BDRVRBDState {
  98     rados_t cluster;
  99     rados_ioctx_t io_ctx;
 100     rbd_image_t image;
 101     char name[RBD_MAX_IMAGE_NAME_SIZE];
 102     char *snap;
 103 } BDRVRBDState;
 104
 105 static int qemu_rbd_next_tok(char *dst, int dst_len,
 106                              char *src, char delim,
 107                              const char *name,
 108                              char **p, Error **errp)
 109 {
 110     int l;
 111     char *end;
 112
 113     *p = NULL;
 114
 115     if (delim != '\0') {
 116         for (end = src; *end; ++end) {
 117             if (*end == delim) {
 118                 break;
 119             }
 120             if (*end == '\\' && end[1] != '\0') {
 121                 end++;
 122             }
 123         }
 124         if (*end == delim) {
 125             *p = end + 1;
 126             *end = '\0';
 127         }
 128     }
 129     l = strlen(src);
 130     if (l >= dst_len) {
 131         error_setg(errp, "%s too long", name);
 132         return -EINVAL;
 133     } else if (l == 0) {
 134         error_setg(errp, "%s too short", name);
 135         return -EINVAL;
 136     }
 137
 138     pstrcpy(dst, dst_len, src);
 139
 140     return 0;
 141 }
 142
 143 static void qemu_rbd_unescape(char *src)
 144 {
 145     char *p;
 146
 147     for (p = src; *src; ++src, ++p) {
 148         if (*src == '\\' && src[1] != '\0') {
 149             src++;
 150         }
 151         *p = *src;
 152     }
 153     *p = '\0';
 154 }
 155
 156 static int qemu_rbd_parsename(const char *filename,
 157                               char *pool, int pool_len,
 158                               char *snap, int snap_len,
 159                               char *name, int name_len,
 160                               char *conf, int conf_len,
 161                               Error **errp)
 162 {
 163     const char *start;
 164     char *p, *buf;
 165     int ret;
 166
 167     if (!strstart(filename, "rbd:", &start)) {
 168         error_setg(errp, "File name must start with 'rbd:'");
 169         return -EINVAL;
 170     }
 171
 172     buf = g_strdup(start);
 173     p = buf;
 174     *snap = '\0';
 175     *conf = '\0';
 176
 177     ret = qemu_rbd_next_tok(pool, pool_len, p,
 178                             '/', "pool name", &p, errp);
 179     if (ret < 0 || !p) {
 180         ret = -EINVAL;
 181         goto done;
 182     }
 183     qemu_rbd_unescape(pool);
 184
 185     if (strchr(p, '@')) {
 186         ret = qemu_rbd_next_tok(name, name_len, p,
 187                                 '@', "object name", &p, errp);
 188         if (ret < 0) {
 189             goto done;
 190         }
 191         ret = qemu_rbd_next_tok(snap, snap_len, p,
 192                                 ':', "snap name", &p, errp);
 193         qemu_rbd_unescape(snap);
 194     } else {
 195         ret = qemu_rbd_next_tok(name, name_len, p,
 196                                 ':', "object name", &p, errp);
 197     }
 198     qemu_rbd_unescape(name);
 199     if (ret < 0 || !p) {
 200         goto done;
 201     }
 202
 203     ret = qemu_rbd_next_tok(conf, conf_len, p,
 204                             '\0', "configuration", &p, errp);
 205
 206 done:
 207     g_free(buf);
 208     return ret;
 209 }
 210
 211 static char *qemu_rbd_parse_clientname(const char *conf, char *clientname)
 212 {
 213     const char *p = conf;
 214
 215     while (*p) {
 216         int len;
 217         const char *end = strchr(p, ':');
 218
 219         if (end) {
 220             len = end - p;
 221         } else {
 222             len = strlen(p);
 223         }
 224
 225         if (strncmp(p, "id=", 3) == 0) {
 226             len -= 3;
 227             strncpy(clientname, p + 3, len);
 228             clientname[len] = '\0';
 229             return clientname;
 230         }
 231         if (end == NULL) {
 232             break;
 233         }
 234         p = end + 1;
 235     }
 236     return NULL;
 237 }
 238
 239
 240 static int qemu_rbd_set_auth(rados_t cluster, const char *secretid,
 241                              Error **errp)
 242 {
 243     if (secretid == 0) {
 244         return 0;
 245     }
 246
 247     gchar *secret = qcrypto_secret_lookup_as_base64(secretid,
 248                                                     errp);
 249     if (!secret) {
 250         return -1;
 251     }
 252
 253     rados_conf_set(cluster, "key", secret);
 254     g_free(secret);
 255
 256     return 0;
 257 }
 258
 259
 260 static int qemu_rbd_set_conf(rados_t cluster, const char *conf,
 261                              bool only_read_conf_file,
 262                              Error **errp)
 263 {
 264     char *p, *buf;
 265     char name[RBD_MAX_CONF_NAME_SIZE];
 266     char value[RBD_MAX_CONF_VAL_SIZE];
 267     int ret = 0;
 268
 269     buf = g_strdup(conf);
 270     p = buf;
 271
 272     while (p) {
 273         ret = qemu_rbd_next_tok(name, sizeof(name), p,
 274                                 '=', "conf option name", &p, errp);
 275         if (ret < 0) {
 276             break;
 277         }
 278         qemu_rbd_unescape(name);
 279
 280         if (!p) {
 281             error_setg(errp, "conf option %s has no value", name);
 282             ret = -EINVAL;
 283             break;
 284         }
 285
 286         ret = qemu_rbd_next_tok(value, sizeof(value), p,
 287                                 ':', "conf option value", &p, errp);
 288         if (ret < 0) {
 289             break;
 290         }
 291         qemu_rbd_unescape(value);
 292
 293         if (strcmp(name, "conf") == 0) {
 294             /* read the conf file alone, so it doesn't override more
 295                specific settings for a particular device */
 296             if (only_read_conf_file) {
 297                 ret = rados_conf_read_file(cluster, value);
 298                 if (ret < 0) {
 299                     error_setg_errno(errp, -ret, "error reading conf file %s",
 300                                      value);
 301                     break;
 302                 }
 303             }
 304         } else if (strcmp(name, "id") == 0) {
 305             /* ignore, this is parsed by qemu_rbd_parse_clientname() */
 306         } else if (!only_read_conf_file) {
 307             ret = rados_conf_set(cluster, name, value);
 308             if (ret < 0) {
 309                 error_setg_errno(errp, -ret, "invalid conf option %s", name);
 310                 ret = -EINVAL;
 311                 break;
 312             }
 313         }
 314     }
 315
 316     g_free(buf);
 317     return ret;
 318 }
 319
 320 static void qemu_rbd_memset(RADOSCB *rcb, int64_t offs)
 321 {
 322     if (LIBRBD_USE_IOVEC) {
 323         RBDAIOCB *acb = rcb->acb;
 324         iov_memset(acb->qiov->iov, acb->qiov->niov, offs, 0,
 325                    acb->qiov->size - offs);
 326     } else {
 327         memset(rcb->buf + offs, 0, rcb->size - offs);
 328     }
 329 }
 330
 331 static int qemu_rbd_create(const char *filename, QemuOpts *opts, Error **errp)
 332 {
 333     Error *local_err = NULL;
 334     int64_t bytes = 0;
 335     int64_t objsize;
 336     int obj_order = 0;
 337     char pool[RBD_MAX_POOL_NAME_SIZE];
 338     char name[RBD_MAX_IMAGE_NAME_SIZE];
 339     char snap_buf[RBD_MAX_SNAP_NAME_SIZE];
 340     char conf[RBD_MAX_CONF_SIZE];
 341     char clientname_buf[RBD_MAX_CONF_SIZE];
 342     char *clientname;
 343     const char *secretid;
 344     rados_t cluster;
 345     rados_ioctx_t io_ctx;
 346     int ret;
 347
 348     secretid = qemu_opt_get(opts, "password-secret");
 349
 350     if (qemu_rbd_parsename(filename, pool, sizeof(pool),
 351                            snap_buf, sizeof(snap_buf),
 352                            name, sizeof(name),
 353                            conf, sizeof(conf), &local_err) < 0) {
 354         error_propagate(errp, local_err);
 355         return -EINVAL;
 356     }
 357
 358     /* Read out options */
 359     bytes = ROUND_UP(qemu_opt_get_size_del(opts, BLOCK_OPT_SIZE, 0),
 360                      BDRV_SECTOR_SIZE);
 361     objsize = qemu_opt_get_size_del(opts, BLOCK_OPT_CLUSTER_SIZE, 0);
 362     if (objsize) {
 363         if ((objsize - 1) & objsize) {    /* not a power of 2? */
 364             error_setg(errp, "obj size needs to be power of 2");
 365             return -EINVAL;
 366         }
 367         if (objsize < 4096) {
 368             error_setg(errp, "obj size too small");
 369             return -EINVAL;
 370         }
 371         obj_order = ctz32(objsize);
 372     }
 373
 374     clientname = qemu_rbd_parse_clientname(conf, clientname_buf);
 375     ret = rados_create(&cluster, clientname);
 376     if (ret < 0) {
 377         error_setg_errno(errp, -ret, "error initializing");
 378         return ret;
 379     }
 380
 381     if (strstr(conf, "conf=") == NULL) {
 382         /* try default location, but ignore failure */
 383         rados_conf_read_file(cluster, NULL);
 384     } else if (conf[0] != '\0' &&
 385                qemu_rbd_set_conf(cluster, conf, true, &local_err) < 0) {
 386         error_propagate(errp, local_err);
 387         ret = -EIO;
 388         goto shutdown;
 389     }
 390
 391     if (conf[0] != '\0' &&
 392         qemu_rbd_set_conf(cluster, conf, false, &local_err) < 0) {
 393         error_propagate(errp, local_err);
 394         ret = -EIO;
 395         goto shutdown;
 396     }
 397
 398     if (qemu_rbd_set_auth(cluster, secretid, errp) < 0) {
 399         ret = -EIO;
 400         goto shutdown;
 401     }
 402
 403     ret = rados_connect(cluster);
 404     if (ret < 0) {
 405         error_setg_errno(errp, -ret, "error connecting");
 406         goto shutdown;
 407     }
 408
 409     ret = rados_ioctx_create(cluster, pool, &io_ctx);
 410     if (ret < 0) {
 411         error_setg_errno(errp, -ret, "error opening pool %s", pool);
 412         goto shutdown;
 413     }
 414
 415     ret = rbd_create(io_ctx, name, bytes, &obj_order);
 416     if (ret < 0) {
 417         error_setg_errno(errp, -ret, "error rbd create");
 418     }
 419
 420     rados_ioctx_destroy(io_ctx);
 421
 422 shutdown:
 423     rados_shutdown(cluster);
 424     return ret;
 425 }
 426
 427 /*
 428  * This aio completion is being called from rbd_finish_bh() and runs in qemu
 429  * BH context.
 430  */
 431 static void qemu_rbd_complete_aio(RADOSCB *rcb)
 432 {
 433     RBDAIOCB *acb = rcb->acb;
 434     int64_t r;
 435
 436     r = rcb->ret;
 437
 438     if (acb->cmd != RBD_AIO_READ) {
 439         if (r < 0) {
 440             acb->ret = r;
 441             acb->error = 1;
 442         } else if (!acb->error) {
 443             acb->ret = rcb->size;
 444         }
 445     } else {
 446         if (r < 0) {
 447             qemu_rbd_memset(rcb, 0);
 448             acb->ret = r;
 449             acb->error = 1;
 450         } else if (r < rcb->size) {
 451             qemu_rbd_memset(rcb, r);
 452             if (!acb->error) {
 453                 acb->ret = rcb->size;
 454             }
 455         } else if (!acb->error) {
 456             acb->ret = r;
 457         }
 458     }
 459
 460     g_free(rcb);
 461
 462     if (!LIBRBD_USE_IOVEC) {
 463         if (acb->cmd == RBD_AIO_READ) {
 464             qemu_iovec_from_buf(acb->qiov, 0, acb->bounce, acb->qiov->size);
 465         }
 466         qemu_vfree(acb->bounce);
 467     }
 468
 469     acb->common.cb(acb->common.opaque, (acb->ret > 0 ? 0 : acb->ret));
 470
 471     qemu_aio_unref(acb);
 472 }
 473
 474 /* TODO Convert to fine grained options */
 475 static QemuOptsList runtime_opts = {
 476     .name = "rbd",
 477     .head = QTAILQ_HEAD_INITIALIZER(runtime_opts.head),
 478     .desc = {
 479         {
 480             .name = "filename",
 481             .type = QEMU_OPT_STRING,
 482             .help = "Specification of the rbd image",
 483         },
 484         {
 485             .name = "password-secret",
 486             .type = QEMU_OPT_STRING,
 487             .help = "ID of secret providing the password",
 488         },
 489         { /* end of list */ }
 490     },
 491 };
 492
 493 static int qemu_rbd_open(BlockDriverState *bs, QDict *options, int flags,
 494                          Error **errp)
 495 {
 496     BDRVRBDState *s = bs->opaque;
 497     char pool[RBD_MAX_POOL_NAME_SIZE];
 498     char snap_buf[RBD_MAX_SNAP_NAME_SIZE];
 499     char conf[RBD_MAX_CONF_SIZE];
 500     char clientname_buf[RBD_MAX_CONF_SIZE];
 501     char *clientname;
 502     const char *secretid;
 503     QemuOpts *opts;
 504     Error *local_err = NULL;
 505     const char *filename;
 506     int r;
 507
 508     opts = qemu_opts_create(&runtime_opts, NULL, 0, &error_abort);
 509     qemu_opts_absorb_qdict(opts, options, &local_err);
 510     if (local_err) {
 511         error_propagate(errp, local_err);
 512         qemu_opts_del(opts);
 513         return -EINVAL;
 514     }
 515
 516     filename = qemu_opt_get(opts, "filename");
 517     secretid = qemu_opt_get(opts, "password-secret");
 518
 519     if (qemu_rbd_parsename(filename, pool, sizeof(pool),
 520                            snap_buf, sizeof(snap_buf),
 521                            s->name, sizeof(s->name),
 522                            conf, sizeof(conf), errp) < 0) {
 523         r = -EINVAL;
 524         goto failed_opts;
 525     }
 526
 527     clientname = qemu_rbd_parse_clientname(conf, clientname_buf);
 528     r = rados_create(&s->cluster, clientname);
 529     if (r < 0) {
 530         error_setg_errno(errp, -r, "error initializing");
 531         goto failed_opts;
 532     }
 533
 534     s->snap = NULL;
 535     if (snap_buf[0] != '\0') {
 536         s->snap = g_strdup(snap_buf);
 537     }
 538
 539     if (strstr(conf, "conf=") == NULL) {
 540         /* try default location, but ignore failure */
 541         rados_conf_read_file(s->cluster, NULL);
 542     } else if (conf[0] != '\0') {
 543         r = qemu_rbd_set_conf(s->cluster, conf, true, errp);
 544         if (r < 0) {
 545             goto failed_shutdown;
 546         }
 547     }
 548
 549     if (conf[0] != '\0') {
 550         r = qemu_rbd_set_conf(s->cluster, conf, false, errp);
 551         if (r < 0) {
 552             goto failed_shutdown;
 553         }
 554     }
 555
 556     if (qemu_rbd_set_auth(s->cluster, secretid, errp) < 0) {
 557         r = -EIO;
 558         goto failed_shutdown;
 559     }
 560
 561     /*
 562      * Fallback to more conservative semantics if setting cache
 563      * options fails. Ignore errors from setting rbd_cache because the
 564      * only possible error is that the option does not exist, and
 565      * librbd defaults to no caching. If write through caching cannot
 566      * be set up, fall back to no caching.
 567      */
 568     if (flags & BDRV_O_NOCACHE) {
 569         rados_conf_set(s->cluster, "rbd_cache", "false");
 570     } else {
 571         rados_conf_set(s->cluster, "rbd_cache", "true");
 572     }
 573
 574     r = rados_connect(s->cluster);
 575     if (r < 0) {
 576         error_setg_errno(errp, -r, "error connecting");
 577         goto failed_shutdown;
 578     }
 579
 580     r = rados_ioctx_create(s->cluster, pool, &s->io_ctx);
 581     if (r < 0) {
 582         error_setg_errno(errp, -r, "error opening pool %s", pool);
 583         goto failed_shutdown;
 584     }
 585
 586     r = rbd_open(s->io_ctx, s->name, &s->image, s->snap);
 587     if (r < 0) {
 588         error_setg_errno(errp, -r, "error reading header from %s", s->name);
 589         goto failed_open;
 590     }
 591
 592     bs->read_only = (s->snap != NULL);
 593
 594     qemu_opts_del(opts);
 595     return 0;
 596
 597 failed_open:
 598     rados_ioctx_destroy(s->io_ctx);
 599 failed_shutdown:
 600     rados_shutdown(s->cluster);
 601     g_free(s->snap);
 602 failed_opts:
 603     qemu_opts_del(opts);
 604     return r;
 605 }
 606
 607 static void qemu_rbd_close(BlockDriverState *bs)
 608 {
 609     BDRVRBDState *s = bs->opaque;
 610
 611     rbd_close(s->image);
 612     rados_ioctx_destroy(s->io_ctx);
 613     g_free(s->snap);
 614     rados_shutdown(s->cluster);
 615 }
 616
 617 static const AIOCBInfo rbd_aiocb_info = {
 618     .aiocb_size = sizeof(RBDAIOCB),
 619 };
 620
 621 static void rbd_finish_bh(void *opaque)
 622 {
 623     RADOSCB *rcb = opaque;
 624     qemu_rbd_complete_aio(rcb);
 625 }
 626
 627 /*
 628  * This is the callback function for rbd_aio_read and _write
 629  *
 630  * Note: this function is being called from a non qemu thread so
 631  * we need to be careful about what we do here. Generally we only
 632  * schedule a BH, and do the rest of the io completion handling
 633  * from rbd_finish_bh() which runs in a qemu context.
 634  */
 635 static void rbd_finish_aiocb(rbd_completion_t c, RADOSCB *rcb)
 636 {
 637     RBDAIOCB *acb = rcb->acb;
 638
 639     rcb->ret = rbd_aio_get_return_value(c);
 640     rbd_aio_release(c);
 641
 642     aio_bh_schedule_oneshot(bdrv_get_aio_context(acb->common.bs),
 643                             rbd_finish_bh, rcb);
 644 }
 645
 646 static int rbd_aio_discard_wrapper(rbd_image_t image,
 647                                    uint64_t off,
 648                                    uint64_t len,
 649                                    rbd_completion_t comp)
 650 {
 651 #ifdef LIBRBD_SUPPORTS_DISCARD
 652     return rbd_aio_discard(image, off, len, comp);
 653 #else
 654     return -ENOTSUP;
 655 #endif
 656 }
 657
 658 static int rbd_aio_flush_wrapper(rbd_image_t image,
 659                                  rbd_completion_t comp)
 660 {
 661 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
 662     return rbd_aio_flush(image, comp);
 663 #else
 664     return -ENOTSUP;
 665 #endif
 666 }
 667
 668 static BlockAIOCB *rbd_start_aio(BlockDriverState *bs,
 669                                  int64_t off,
 670                                  QEMUIOVector *qiov,
 671                                  int64_t size,
 672                                  BlockCompletionFunc *cb,
 673                                  void *opaque,
 674                                  RBDAIOCmd cmd)
 675 {
 676     RBDAIOCB *acb;
 677     RADOSCB *rcb = NULL;
 678     rbd_completion_t c;
 679     int r;
 680
 681     BDRVRBDState *s = bs->opaque;
 682
 683     acb = qemu_aio_get(&rbd_aiocb_info, bs, cb, opaque);
 684     acb->cmd = cmd;
 685     acb->qiov = qiov;
 686     assert(!qiov || qiov->size == size);
 687
 688     rcb = g_new(RADOSCB, 1);
 689
 690     if (!LIBRBD_USE_IOVEC) {
 691         if (cmd == RBD_AIO_DISCARD || cmd == RBD_AIO_FLUSH) {
 692             acb->bounce = NULL;
 693         } else {
 694             acb->bounce = qemu_try_blockalign(bs, qiov->size);
 695             if (acb->bounce == NULL) {
 696                 goto failed;
 697             }
 698         }
 699         if (cmd == RBD_AIO_WRITE) {
 700             qemu_iovec_to_buf(acb->qiov, 0, acb->bounce, qiov->size);
 701         }
 702         rcb->buf = acb->bounce;
 703     }
 704
 705     acb->ret = 0;
 706     acb->error = 0;
 707     acb->s = s;
 708
 709     rcb->acb = acb;
 710     rcb->s = acb->s;
 711     rcb->size = size;
 712     r = rbd_aio_create_completion(rcb, (rbd_callback_t) rbd_finish_aiocb, &c);
 713     if (r < 0) {
 714         goto failed;
 715     }
 716
 717     switch (cmd) {
 718     case RBD_AIO_WRITE:
 719 #ifdef LIBRBD_SUPPORTS_IOVEC
 720             r = rbd_aio_writev(s->image, qiov->iov, qiov->niov, off, c);
 721 #else
 722             r = rbd_aio_write(s->image, off, size, rcb->buf, c);
 723 #endif
 724         break;
 725     case RBD_AIO_READ:
 726 #ifdef LIBRBD_SUPPORTS_IOVEC
 727             r = rbd_aio_readv(s->image, qiov->iov, qiov->niov, off, c);
 728 #else
 729             r = rbd_aio_read(s->image, off, size, rcb->buf, c);
 730 #endif
 731         break;
 732     case RBD_AIO_DISCARD:
 733         r = rbd_aio_discard_wrapper(s->image, off, size, c);
 734         break;
 735     case RBD_AIO_FLUSH:
 736         r = rbd_aio_flush_wrapper(s->image, c);
 737         break;
 738     default:
 739         r = -EINVAL;
 740     }
 741
 742     if (r < 0) {
 743         goto failed_completion;
 744     }
 745     return &acb->common;
 746
 747 failed_completion:
 748     rbd_aio_release(c);
 749 failed:
 750     g_free(rcb);
 751     if (!LIBRBD_USE_IOVEC) {
 752         qemu_vfree(acb->bounce);
 753     }
 754
 755     qemu_aio_unref(acb);
 756     return NULL;
 757 }
 758
 759 static BlockAIOCB *qemu_rbd_aio_readv(BlockDriverState *bs,
 760                                       int64_t sector_num,
 761                                       QEMUIOVector *qiov,
 762                                       int nb_sectors,
 763                                       BlockCompletionFunc *cb,
 764                                       void *opaque)
 765 {
 766     return rbd_start_aio(bs, sector_num << BDRV_SECTOR_BITS, qiov,
 767                          (int64_t) nb_sectors << BDRV_SECTOR_BITS, cb, opaque,
 768                          RBD_AIO_READ);
 769 }
 770
 771 static BlockAIOCB *qemu_rbd_aio_writev(BlockDriverState *bs,
 772                                        int64_t sector_num,
 773                                        QEMUIOVector *qiov,
 774                                        int nb_sectors,
 775                                        BlockCompletionFunc *cb,
 776                                        void *opaque)
 777 {
 778     return rbd_start_aio(bs, sector_num << BDRV_SECTOR_BITS, qiov,
 779                          (int64_t) nb_sectors << BDRV_SECTOR_BITS, cb, opaque,
 780                          RBD_AIO_WRITE);
 781 }
 782
 783 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
 784 static BlockAIOCB *qemu_rbd_aio_flush(BlockDriverState *bs,
 785                                       BlockCompletionFunc *cb,
 786                                       void *opaque)
 787 {
 788     return rbd_start_aio(bs, 0, NULL, 0, cb, opaque, RBD_AIO_FLUSH);
 789 }
 790
 791 #else
 792
 793 static int qemu_rbd_co_flush(BlockDriverState *bs)
 794 {
 795 #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 1)
 796     /* rbd_flush added in 0.1.1 */
 797     BDRVRBDState *s = bs->opaque;
 798     return rbd_flush(s->image);
 799 #else
 800     return 0;
 801 #endif
 802 }
 803 #endif
 804
 805 static int qemu_rbd_getinfo(BlockDriverState *bs, BlockDriverInfo *bdi)
 806 {
 807     BDRVRBDState *s = bs->opaque;
 808     rbd_image_info_t info;
 809     int r;
 810
 811     r = rbd_stat(s->image, &info, sizeof(info));
 812     if (r < 0) {
 813         return r;
 814     }
 815
 816     bdi->cluster_size = info.obj_size;
 817     return 0;
 818 }
 819
 820 static int64_t qemu_rbd_getlength(BlockDriverState *bs)
 821 {
 822     BDRVRBDState *s = bs->opaque;
 823     rbd_image_info_t info;
 824     int r;
 825
 826     r = rbd_stat(s->image, &info, sizeof(info));
 827     if (r < 0) {
 828         return r;
 829     }
 830
 831     return info.size;
 832 }
 833
 834 static int qemu_rbd_truncate(BlockDriverState *bs, int64_t offset)
 835 {
 836     BDRVRBDState *s = bs->opaque;
 837     int r;
 838
 839     r = rbd_resize(s->image, offset);
 840     if (r < 0) {
 841         return r;
 842     }
 843
 844     return 0;
 845 }
 846
 847 static int qemu_rbd_snap_create(BlockDriverState *bs,
 848                                 QEMUSnapshotInfo *sn_info)
 849 {
 850     BDRVRBDState *s = bs->opaque;
 851     int r;
 852
 853     if (sn_info->name[0] == '\0') {
 854         return -EINVAL; /* we need a name for rbd snapshots */
 855     }
 856
 857     /*
 858      * rbd snapshots are using the name as the user controlled unique identifier
 859      * we can't use the rbd snapid for that purpose, as it can't be set
 860      */
 861     if (sn_info->id_str[0] != '\0' &&
 862         strcmp(sn_info->id_str, sn_info->name) != 0) {
 863         return -EINVAL;
 864     }
 865
 866     if (strlen(sn_info->name) >= sizeof(sn_info->id_str)) {
 867         return -ERANGE;
 868     }
 869
 870     r = rbd_snap_create(s->image, sn_info->name);
 871     if (r < 0) {
 872         error_report("failed to create snap: %s", strerror(-r));
 873         return r;
 874     }
 875
 876     return 0;
 877 }
 878
 879 static int qemu_rbd_snap_remove(BlockDriverState *bs,
 880                                 const char *snapshot_id,
 881                                 const char *snapshot_name,
 882                                 Error **errp)
 883 {
 884     BDRVRBDState *s = bs->opaque;
 885     int r;
 886
 887     if (!snapshot_name) {
 888         error_setg(errp, "rbd need a valid snapshot name");
 889         return -EINVAL;
 890     }
 891
 892     /* If snapshot_id is specified, it must be equal to name, see
 893        qemu_rbd_snap_list() */
 894     if (snapshot_id && strcmp(snapshot_id, snapshot_name)) {
 895         error_setg(errp,
 896                    "rbd do not support snapshot id, it should be NULL or "
 897                    "equal to snapshot name");
 898         return -EINVAL;
 899     }
 900
 901     r = rbd_snap_remove(s->image, snapshot_name);
 902     if (r < 0) {
 903         error_setg_errno(errp, -r, "Failed to remove the snapshot");
 904     }
 905     return r;
 906 }
 907
 908 static int qemu_rbd_snap_rollback(BlockDriverState *bs,
 909                                   const char *snapshot_name)
 910 {
 911     BDRVRBDState *s = bs->opaque;
 912
 913     return rbd_snap_rollback(s->image, snapshot_name);
 914 }
 915
 916 static int qemu_rbd_snap_list(BlockDriverState *bs,
 917                               QEMUSnapshotInfo **psn_tab)
 918 {
 919     BDRVRBDState *s = bs->opaque;
 920     QEMUSnapshotInfo *sn_info, *sn_tab = NULL;
 921     int i, snap_count;
 922     rbd_snap_info_t *snaps;
 923     int max_snaps = RBD_MAX_SNAPS;
 924
 925     do {
 926         snaps = g_new(rbd_snap_info_t, max_snaps);
 927         snap_count = rbd_snap_list(s->image, snaps, &max_snaps);
 928         if (snap_count <= 0) {
 929             g_free(snaps);
 930         }
 931     } while (snap_count == -ERANGE);
 932
 933     if (snap_count <= 0) {
 934         goto done;
 935     }
 936
 937     sn_tab = g_new0(QEMUSnapshotInfo, snap_count);
 938
 939     for (i = 0; i < snap_count; i++) {
 940         const char *snap_name = snaps[i].name;
 941
 942         sn_info = sn_tab + i;
 943         pstrcpy(sn_info->id_str, sizeof(sn_info->id_str), snap_name);
 944         pstrcpy(sn_info->name, sizeof(sn_info->name), snap_name);
 945
 946         sn_info->vm_state_size = snaps[i].size;
 947         sn_info->date_sec = 0;
 948         sn_info->date_nsec = 0;
 949         sn_info->vm_clock_nsec = 0;
 950     }
 951     rbd_snap_list_end(snaps);
 952     g_free(snaps);
 953
 954  done:
 955     *psn_tab = sn_tab;
 956     return snap_count;
 957 }
 958
 959 #ifdef LIBRBD_SUPPORTS_DISCARD
 960 static BlockAIOCB *qemu_rbd_aio_pdiscard(BlockDriverState *bs,
 961                                          int64_t offset,
 962                                          int count,
 963                                          BlockCompletionFunc *cb,
 964                                          void *opaque)
 965 {
 966     return rbd_start_aio(bs, offset, NULL, count, cb, opaque,
 967                          RBD_AIO_DISCARD);
 968 }
 969 #endif
 970
 971 #ifdef LIBRBD_SUPPORTS_INVALIDATE
 972 static void qemu_rbd_invalidate_cache(BlockDriverState *bs,
 973                                       Error **errp)
 974 {
 975     BDRVRBDState *s = bs->opaque;
 976     int r = rbd_invalidate_cache(s->image);
 977     if (r < 0) {
 978         error_setg_errno(errp, -r, "Failed to invalidate the cache");
 979     }
 980 }
 981 #endif
 982
 983 static QemuOptsList qemu_rbd_create_opts = {
 984     .name = "rbd-create-opts",
 985     .head = QTAILQ_HEAD_INITIALIZER(qemu_rbd_create_opts.head),
 986     .desc = {
 987         {
 988             .name = BLOCK_OPT_SIZE,
 989             .type = QEMU_OPT_SIZE,
 990             .help = "Virtual disk size"
 991         },
 992         {
 993             .name = BLOCK_OPT_CLUSTER_SIZE,
 994             .type = QEMU_OPT_SIZE,
 995             .help = "RBD object size"
 996         },
 997         {
 998             .name = "password-secret",
 999             .type = QEMU_OPT_STRING,
1000             .help = "ID of secret providing the password",
1001         },
1002         { /* end of list */ }
1003     }
1004 };
1005
1006 static BlockDriver bdrv_rbd = {
1007     .format_name        = "rbd",
1008     .instance_size      = sizeof(BDRVRBDState),
1009     .bdrv_needs_filename = true,
1010     .bdrv_file_open     = qemu_rbd_open,
1011     .bdrv_close         = qemu_rbd_close,
1012     .bdrv_create        = qemu_rbd_create,
1013     .bdrv_has_zero_init = bdrv_has_zero_init_1,
1014     .bdrv_get_info      = qemu_rbd_getinfo,
1015     .create_opts        = &qemu_rbd_create_opts,
1016     .bdrv_getlength     = qemu_rbd_getlength,
1017     .bdrv_truncate      = qemu_rbd_truncate,
1018     .protocol_name      = "rbd",
1019
1020     .bdrv_aio_readv         = qemu_rbd_aio_readv,
1021     .bdrv_aio_writev        = qemu_rbd_aio_writev,
1022
1023 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
1024     .bdrv_aio_flush         = qemu_rbd_aio_flush,
1025 #else
1026     .bdrv_co_flush_to_disk  = qemu_rbd_co_flush,
1027 #endif
1028
1029 #ifdef LIBRBD_SUPPORTS_DISCARD
1030     .bdrv_aio_pdiscard      = qemu_rbd_aio_pdiscard,
1031 #endif
1032
1033     .bdrv_snapshot_create   = qemu_rbd_snap_create,
1034     .bdrv_snapshot_delete   = qemu_rbd_snap_remove,
1035     .bdrv_snapshot_list     = qemu_rbd_snap_list,
1036     .bdrv_snapshot_goto     = qemu_rbd_snap_rollback,
1037 #ifdef LIBRBD_SUPPORTS_INVALIDATE
1038     .bdrv_invalidate_cache  = qemu_rbd_invalidate_cache,
1039 #endif
1040 };
1041
1042 static void bdrv_rbd_init(void)
1043 {
1044     bdrv_register(&bdrv_rbd);
1045 }
1046
1047 block_init(bdrv_rbd_init);