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  */
  11
  12 #include <inttypes.h>
  13
  14 #include "qemu-common.h"
  15 #include "qemu-error.h"
  16
  17 #include "block_int.h"
  18
  19 #include <rbd/librbd.h>
  20
  21
  22
  23 /*
  24  * When specifying the image filename use:
  25  *
  26  * rbd:poolname/devicename[@snapshotname][:option1=value1[:option2=value2...]]
  27  *
  28  * poolname must be the name of an existing rados pool
  29  *
  30  * devicename is the basename for all objects used to
  31  * emulate the raw device.
  32  *
  33  * Each option given is used to configure rados, and may be
  34  * any Ceph option, or "conf". The "conf" option specifies
  35  * a Ceph configuration file to read.
  36  *
  37  * Metadata information (image size, ...) is stored in an
  38  * object with the name "devicename.rbd".
  39  *
  40  * The raw device is split into 4MB sized objects by default.
  41  * The sequencenumber is encoded in a 12 byte long hex-string,
  42  * and is attached to the devicename, separated by a dot.
  43  * e.g. "devicename.1234567890ab"
  44  *
  45  */
  46
  47 #define OBJ_MAX_SIZE (1UL << OBJ_DEFAULT_OBJ_ORDER)
  48
  49 #define RBD_MAX_CONF_NAME_SIZE 128
  50 #define RBD_MAX_CONF_VAL_SIZE 512
  51 #define RBD_MAX_CONF_SIZE 1024
  52 #define RBD_MAX_POOL_NAME_SIZE 128
  53 #define RBD_MAX_SNAP_NAME_SIZE 128
  54 #define RBD_MAX_SNAPS 100
  55
  56 typedef struct RBDAIOCB {
  57     BlockDriverAIOCB common;
  58     QEMUBH *bh;
  59     int ret;
  60     QEMUIOVector *qiov;
  61     char *bounce;
  62     int write;
  63     int64_t sector_num;
  64     int error;
  65     struct BDRVRBDState *s;
  66     int cancelled;
  67 } RBDAIOCB;
  68
  69 typedef struct RADOSCB {
  70     int rcbid;
  71     RBDAIOCB *acb;
  72     struct BDRVRBDState *s;
  73     int done;
  74     int64_t size;
  75     char *buf;
  76     int ret;
  77 } RADOSCB;
  78
  79 #define RBD_FD_READ 0
  80 #define RBD_FD_WRITE 1
  81
  82 typedef struct BDRVRBDState {
  83     int fds[2];
  84     rados_t cluster;
  85     rados_ioctx_t io_ctx;
  86     rbd_image_t image;
  87     char name[RBD_MAX_IMAGE_NAME_SIZE];
  88     int qemu_aio_count;
  89     char *snap;
  90     int event_reader_pos;
  91     RADOSCB *event_rcb;
  92 } BDRVRBDState;
  93
  94 static void rbd_aio_bh_cb(void *opaque);
  95
  96 static int qemu_rbd_next_tok(char *dst, int dst_len,
  97                              char *src, char delim,
  98                              const char *name,
  99                              char **p)
 100 {
 101     int l;
 102     char *end;
 103
 104     *p = NULL;
 105
 106     if (delim != '\0') {
 107         end = strchr(src, delim);
 108         if (end) {
 109             *p = end + 1;
 110             *end = '\0';
 111         }
 112     }
 113     l = strlen(src);
 114     if (l >= dst_len) {
 115         error_report("%s too long", name);
 116         return -EINVAL;
 117     } else if (l == 0) {
 118         error_report("%s too short", name);
 119         return -EINVAL;
 120     }
 121
 122     pstrcpy(dst, dst_len, src);
 123
 124     return 0;
 125 }
 126
 127 static int qemu_rbd_parsename(const char *filename,
 128                               char *pool, int pool_len,
 129                               char *snap, int snap_len,
 130                               char *name, int name_len,
 131                               char *conf, int conf_len)
 132 {
 133     const char *start;
 134     char *p, *buf;
 135     int ret;
 136
 137     if (!strstart(filename, "rbd:", &start)) {
 138         return -EINVAL;
 139     }
 140
 141     buf = g_strdup(start);
 142     p = buf;
 143     *snap = '\0';
 144     *conf = '\0';
 145
 146     ret = qemu_rbd_next_tok(pool, pool_len, p, '/', "pool name", &p);
 147     if (ret < 0 || !p) {
 148         ret = -EINVAL;
 149         goto done;
 150     }
 151
 152     if (strchr(p, '@')) {
 153         ret = qemu_rbd_next_tok(name, name_len, p, '@', "object name", &p);
 154         if (ret < 0) {
 155             goto done;
 156         }
 157         ret = qemu_rbd_next_tok(snap, snap_len, p, ':', "snap name", &p);
 158     } else {
 159         ret = qemu_rbd_next_tok(name, name_len, p, ':', "object name", &p);
 160     }
 161     if (ret < 0 || !p) {
 162         goto done;
 163     }
 164
 165     ret = qemu_rbd_next_tok(conf, conf_len, p, '\0', "configuration", &p);
 166
 167 done:
 168     g_free(buf);
 169     return ret;
 170 }
 171
 172 static char *qemu_rbd_parse_clientname(const char *conf, char *clientname)
 173 {
 174     const char *p = conf;
 175
 176     while (*p) {
 177         int len;
 178         const char *end = strchr(p, ':');
 179
 180         if (end) {
 181             len = end - p;
 182         } else {
 183             len = strlen(p);
 184         }
 185
 186         if (strncmp(p, "id=", 3) == 0) {
 187             len -= 3;
 188             strncpy(clientname, p + 3, len);
 189             clientname[len] = '\0';
 190             return clientname;
 191         }
 192         if (end == NULL) {
 193             break;
 194         }
 195         p = end + 1;
 196     }
 197     return NULL;
 198 }
 199
 200 static int qemu_rbd_set_conf(rados_t cluster, const char *conf)
 201 {
 202     char *p, *buf;
 203     char name[RBD_MAX_CONF_NAME_SIZE];
 204     char value[RBD_MAX_CONF_VAL_SIZE];
 205     int ret = 0;
 206
 207     buf = g_strdup(conf);
 208     p = buf;
 209
 210     while (p) {
 211         ret = qemu_rbd_next_tok(name, sizeof(name), p,
 212                                 '=', "conf option name", &p);
 213         if (ret < 0) {
 214             break;
 215         }
 216
 217         if (!p) {
 218             error_report("conf option %s has no value", name);
 219             ret = -EINVAL;
 220             break;
 221         }
 222
 223         ret = qemu_rbd_next_tok(value, sizeof(value), p,
 224                                 ':', "conf option value", &p);
 225         if (ret < 0) {
 226             break;
 227         }
 228
 229         if (strcmp(name, "conf") == 0) {
 230             ret = rados_conf_read_file(cluster, value);
 231             if (ret < 0) {
 232                 error_report("error reading conf file %s", value);
 233                 break;
 234             }
 235         } else if (strcmp(name, "id") == 0) {
 236             /* ignore, this is parsed by qemu_rbd_parse_clientname() */
 237         } else {
 238             ret = rados_conf_set(cluster, name, value);
 239             if (ret < 0) {
 240                 error_report("invalid conf option %s", name);
 241                 ret = -EINVAL;
 242                 break;
 243             }
 244         }
 245     }
 246
 247     g_free(buf);
 248     return ret;
 249 }
 250
 251 static int qemu_rbd_create(const char *filename, QEMUOptionParameter *options)
 252 {
 253     int64_t bytes = 0;
 254     int64_t objsize;
 255     int obj_order = 0;
 256     char pool[RBD_MAX_POOL_NAME_SIZE];
 257     char name[RBD_MAX_IMAGE_NAME_SIZE];
 258     char snap_buf[RBD_MAX_SNAP_NAME_SIZE];
 259     char conf[RBD_MAX_CONF_SIZE];
 260     char clientname_buf[RBD_MAX_CONF_SIZE];
 261     char *clientname;
 262     rados_t cluster;
 263     rados_ioctx_t io_ctx;
 264     int ret;
 265
 266     if (qemu_rbd_parsename(filename, pool, sizeof(pool),
 267                            snap_buf, sizeof(snap_buf),
 268                            name, sizeof(name),
 269                            conf, sizeof(conf)) < 0) {
 270         return -EINVAL;
 271     }
 272
 273     /* Read out options */
 274     while (options && options->name) {
 275         if (!strcmp(options->name, BLOCK_OPT_SIZE)) {
 276             bytes = options->value.n;
 277         } else if (!strcmp(options->name, BLOCK_OPT_CLUSTER_SIZE)) {
 278             if (options->value.n) {
 279                 objsize = options->value.n;
 280                 if ((objsize - 1) & objsize) {    /* not a power of 2? */
 281                     error_report("obj size needs to be power of 2");
 282                     return -EINVAL;
 283                 }
 284                 if (objsize < 4096) {
 285                     error_report("obj size too small");
 286                     return -EINVAL;
 287                 }
 288                 obj_order = ffs(objsize) - 1;
 289             }
 290         }
 291         options++;
 292     }
 293
 294     clientname = qemu_rbd_parse_clientname(conf, clientname_buf);
 295     if (rados_create(&cluster, clientname) < 0) {
 296         error_report("error initializing");
 297         return -EIO;
 298     }
 299
 300     if (strstr(conf, "conf=") == NULL) {
 301         if (rados_conf_read_file(cluster, NULL) < 0) {
 302             error_report("error reading config file");
 303             rados_shutdown(cluster);
 304             return -EIO;
 305         }
 306     }
 307
 308     if (conf[0] != '\0' &&
 309         qemu_rbd_set_conf(cluster, conf) < 0) {
 310         error_report("error setting config options");
 311         rados_shutdown(cluster);
 312         return -EIO;
 313     }
 314
 315     if (rados_connect(cluster) < 0) {
 316         error_report("error connecting");
 317         rados_shutdown(cluster);
 318         return -EIO;
 319     }
 320
 321     if (rados_ioctx_create(cluster, pool, &io_ctx) < 0) {
 322         error_report("error opening pool %s", pool);
 323         rados_shutdown(cluster);
 324         return -EIO;
 325     }
 326
 327     ret = rbd_create(io_ctx, name, bytes, &obj_order);
 328     rados_ioctx_destroy(io_ctx);
 329     rados_shutdown(cluster);
 330
 331     return ret;
 332 }
 333
 334 /*
 335  * This aio completion is being called from qemu_rbd_aio_event_reader()
 336  * and runs in qemu context. It schedules a bh, but just in case the aio
 337  * was not cancelled before.
 338  */
 339 static void qemu_rbd_complete_aio(RADOSCB *rcb)
 340 {
 341     RBDAIOCB *acb = rcb->acb;
 342     int64_t r;
 343
 344     if (acb->cancelled) {
 345         qemu_vfree(acb->bounce);
 346         qemu_aio_release(acb);
 347         goto done;
 348     }
 349
 350     r = rcb->ret;
 351
 352     if (acb->write) {
 353         if (r < 0) {
 354             acb->ret = r;
 355             acb->error = 1;
 356         } else if (!acb->error) {
 357             acb->ret = rcb->size;
 358         }
 359     } else {
 360         if (r < 0) {
 361             memset(rcb->buf, 0, rcb->size);
 362             acb->ret = r;
 363             acb->error = 1;
 364         } else if (r < rcb->size) {
 365             memset(rcb->buf + r, 0, rcb->size - r);
 366             if (!acb->error) {
 367                 acb->ret = rcb->size;
 368             }
 369         } else if (!acb->error) {
 370             acb->ret = r;
 371         }
 372     }
 373     /* Note that acb->bh can be NULL in case where the aio was cancelled */
 374     acb->bh = qemu_bh_new(rbd_aio_bh_cb, acb);
 375     qemu_bh_schedule(acb->bh);
 376 done:
 377     g_free(rcb);
 378 }
 379
 380 /*
 381  * aio fd read handler. It runs in the qemu context and calls the
 382  * completion handling of completed rados aio operations.
 383  */
 384 static void qemu_rbd_aio_event_reader(void *opaque)
 385 {
 386     BDRVRBDState *s = opaque;
 387
 388     ssize_t ret;
 389
 390     do {
 391         char *p = (char *)&s->event_rcb;
 392
 393         /* now read the rcb pointer that was sent from a non qemu thread */
 394         ret = read(s->fds[RBD_FD_READ], p + s->event_reader_pos,
 395                    sizeof(s->event_rcb) - s->event_reader_pos);
 396         if (ret > 0) {
 397             s->event_reader_pos += ret;
 398             if (s->event_reader_pos == sizeof(s->event_rcb)) {
 399                 s->event_reader_pos = 0;
 400                 qemu_rbd_complete_aio(s->event_rcb);
 401                 s->qemu_aio_count--;
 402             }
 403         }
 404     } while (ret < 0 && errno == EINTR);
 405 }
 406
 407 static int qemu_rbd_aio_flush_cb(void *opaque)
 408 {
 409     BDRVRBDState *s = opaque;
 410
 411     return (s->qemu_aio_count > 0);
 412 }
 413
 414 static int qemu_rbd_open(BlockDriverState *bs, const char *filename, int flags)
 415 {
 416     BDRVRBDState *s = bs->opaque;
 417     char pool[RBD_MAX_POOL_NAME_SIZE];
 418     char snap_buf[RBD_MAX_SNAP_NAME_SIZE];
 419     char conf[RBD_MAX_CONF_SIZE];
 420     char clientname_buf[RBD_MAX_CONF_SIZE];
 421     char *clientname;
 422     int r;
 423
 424     if (qemu_rbd_parsename(filename, pool, sizeof(pool),
 425                            snap_buf, sizeof(snap_buf),
 426                            s->name, sizeof(s->name),
 427                            conf, sizeof(conf)) < 0) {
 428         return -EINVAL;
 429     }
 430
 431     clientname = qemu_rbd_parse_clientname(conf, clientname_buf);
 432     r = rados_create(&s->cluster, clientname);
 433     if (r < 0) {
 434         error_report("error initializing");
 435         return r;
 436     }
 437
 438     s->snap = NULL;
 439     if (snap_buf[0] != '\0') {
 440         s->snap = g_strdup(snap_buf);
 441     }
 442
 443     if (strstr(conf, "conf=") == NULL) {
 444         r = rados_conf_read_file(s->cluster, NULL);
 445         if (r < 0) {
 446             error_report("error reading config file");
 447             goto failed_shutdown;
 448         }
 449     }
 450
 451     if (conf[0] != '\0') {
 452         r = qemu_rbd_set_conf(s->cluster, conf);
 453         if (r < 0) {
 454             error_report("error setting config options");
 455             goto failed_shutdown;
 456         }
 457     }
 458
 459     r = rados_connect(s->cluster);
 460     if (r < 0) {
 461         error_report("error connecting");
 462         goto failed_shutdown;
 463     }
 464
 465     r = rados_ioctx_create(s->cluster, pool, &s->io_ctx);
 466     if (r < 0) {
 467         error_report("error opening pool %s", pool);
 468         goto failed_shutdown;
 469     }
 470
 471     r = rbd_open(s->io_ctx, s->name, &s->image, s->snap);
 472     if (r < 0) {
 473         error_report("error reading header from %s", s->name);
 474         goto failed_open;
 475     }
 476
 477     bs->read_only = (s->snap != NULL);
 478
 479     s->event_reader_pos = 0;
 480     r = qemu_pipe(s->fds);
 481     if (r < 0) {
 482         error_report("error opening eventfd");
 483         goto failed;
 484     }
 485     fcntl(s->fds[0], F_SETFL, O_NONBLOCK);
 486     fcntl(s->fds[1], F_SETFL, O_NONBLOCK);
 487     qemu_aio_set_fd_handler(s->fds[RBD_FD_READ], qemu_rbd_aio_event_reader,
 488                             NULL, qemu_rbd_aio_flush_cb, NULL, s);
 489
 490
 491     return 0;
 492
 493 failed:
 494     rbd_close(s->image);
 495 failed_open:
 496     rados_ioctx_destroy(s->io_ctx);
 497 failed_shutdown:
 498     rados_shutdown(s->cluster);
 499     g_free(s->snap);
 500     return r;
 501 }
 502
 503 static void qemu_rbd_close(BlockDriverState *bs)
 504 {
 505     BDRVRBDState *s = bs->opaque;
 506
 507     close(s->fds[0]);
 508     close(s->fds[1]);
 509     qemu_aio_set_fd_handler(s->fds[RBD_FD_READ], NULL , NULL, NULL, NULL,
 510         NULL);
 511
 512     rbd_close(s->image);
 513     rados_ioctx_destroy(s->io_ctx);
 514     g_free(s->snap);
 515     rados_shutdown(s->cluster);
 516 }
 517
 518 /*
 519  * Cancel aio. Since we don't reference acb in a non qemu threads,
 520  * it is safe to access it here.
 521  */
 522 static void qemu_rbd_aio_cancel(BlockDriverAIOCB *blockacb)
 523 {
 524     RBDAIOCB *acb = (RBDAIOCB *) blockacb;
 525     acb->cancelled = 1;
 526 }
 527
 528 static AIOPool rbd_aio_pool = {
 529     .aiocb_size = sizeof(RBDAIOCB),
 530     .cancel = qemu_rbd_aio_cancel,
 531 };
 532
 533 static int qemu_rbd_send_pipe(BDRVRBDState *s, RADOSCB *rcb)
 534 {
 535     int ret = 0;
 536     while (1) {
 537         fd_set wfd;
 538         int fd = s->fds[RBD_FD_WRITE];
 539
 540         /* send the op pointer to the qemu thread that is responsible
 541            for the aio/op completion. Must do it in a qemu thread context */
 542         ret = write(fd, (void *)&rcb, sizeof(rcb));
 543         if (ret >= 0) {
 544             break;
 545         }
 546         if (errno == EINTR) {
 547             continue;
 548         }
 549         if (errno != EAGAIN) {
 550             break;
 551         }
 552
 553         FD_ZERO(&wfd);
 554         FD_SET(fd, &wfd);
 555         do {
 556             ret = select(fd + 1, NULL, &wfd, NULL, NULL);
 557         } while (ret < 0 && errno == EINTR);
 558     }
 559
 560     return ret;
 561 }
 562
 563 /*
 564  * This is the callback function for rbd_aio_read and _write
 565  *
 566  * Note: this function is being called from a non qemu thread so
 567  * we need to be careful about what we do here. Generally we only
 568  * write to the block notification pipe, and do the rest of the
 569  * io completion handling from qemu_rbd_aio_event_reader() which
 570  * runs in a qemu context.
 571  */
 572 static void rbd_finish_aiocb(rbd_completion_t c, RADOSCB *rcb)
 573 {
 574     int ret;
 575     rcb->ret = rbd_aio_get_return_value(c);
 576     rbd_aio_release(c);
 577     ret = qemu_rbd_send_pipe(rcb->s, rcb);
 578     if (ret < 0) {
 579         error_report("failed writing to acb->s->fds");
 580         g_free(rcb);
 581     }
 582 }
 583
 584 /* Callback when all queued rbd_aio requests are complete */
 585
 586 static void rbd_aio_bh_cb(void *opaque)
 587 {
 588     RBDAIOCB *acb = opaque;
 589
 590     if (!acb->write) {
 591         qemu_iovec_from_buffer(acb->qiov, acb->bounce, acb->qiov->size);
 592     }
 593     qemu_vfree(acb->bounce);
 594     acb->common.cb(acb->common.opaque, (acb->ret > 0 ? 0 : acb->ret));
 595     qemu_bh_delete(acb->bh);
 596     acb->bh = NULL;
 597
 598     qemu_aio_release(acb);
 599 }
 600
 601 static BlockDriverAIOCB *rbd_aio_rw_vector(BlockDriverState *bs,
 602                                            int64_t sector_num,
 603                                            QEMUIOVector *qiov,
 604                                            int nb_sectors,
 605                                            BlockDriverCompletionFunc *cb,
 606                                            void *opaque, int write)
 607 {
 608     RBDAIOCB *acb;
 609     RADOSCB *rcb;
 610     rbd_completion_t c;
 611     int64_t off, size;
 612     char *buf;
 613     int r;
 614
 615     BDRVRBDState *s = bs->opaque;
 616
 617     acb = qemu_aio_get(&rbd_aio_pool, bs, cb, opaque);
 618     if (!acb) {
 619         return NULL;
 620     }
 621     acb->write = write;
 622     acb->qiov = qiov;
 623     acb->bounce = qemu_blockalign(bs, qiov->size);
 624     acb->ret = 0;
 625     acb->error = 0;
 626     acb->s = s;
 627     acb->cancelled = 0;
 628     acb->bh = NULL;
 629
 630     if (write) {
 631         qemu_iovec_to_buffer(acb->qiov, acb->bounce);
 632     }
 633
 634     buf = acb->bounce;
 635
 636     off = sector_num * BDRV_SECTOR_SIZE;
 637     size = nb_sectors * BDRV_SECTOR_SIZE;
 638
 639     s->qemu_aio_count++; /* All the RADOSCB */
 640
 641     rcb = g_malloc(sizeof(RADOSCB));
 642     rcb->done = 0;
 643     rcb->acb = acb;
 644     rcb->buf = buf;
 645     rcb->s = acb->s;
 646     rcb->size = size;
 647     r = rbd_aio_create_completion(rcb, (rbd_callback_t) rbd_finish_aiocb, &c);
 648     if (r < 0) {
 649         goto failed;
 650     }
 651
 652     if (write) {
 653         r = rbd_aio_write(s->image, off, size, buf, c);
 654     } else {
 655         r = rbd_aio_read(s->image, off, size, buf, c);
 656     }
 657
 658     if (r < 0) {
 659         goto failed;
 660     }
 661
 662     return &acb->common;
 663
 664 failed:
 665     g_free(rcb);
 666     s->qemu_aio_count--;
 667     qemu_aio_release(acb);
 668     return NULL;
 669 }
 670
 671 static BlockDriverAIOCB *qemu_rbd_aio_readv(BlockDriverState *bs,
 672                                             int64_t sector_num,
 673                                             QEMUIOVector *qiov,
 674                                             int nb_sectors,
 675                                             BlockDriverCompletionFunc *cb,
 676                                             void *opaque)
 677 {
 678     return rbd_aio_rw_vector(bs, sector_num, qiov, nb_sectors, cb, opaque, 0);
 679 }
 680
 681 static BlockDriverAIOCB *qemu_rbd_aio_writev(BlockDriverState *bs,
 682                                              int64_t sector_num,
 683                                              QEMUIOVector *qiov,
 684                                              int nb_sectors,
 685                                              BlockDriverCompletionFunc *cb,
 686                                              void *opaque)
 687 {
 688     return rbd_aio_rw_vector(bs, sector_num, qiov, nb_sectors, cb, opaque, 1);
 689 }
 690
 691 static int qemu_rbd_getinfo(BlockDriverState *bs, BlockDriverInfo *bdi)
 692 {
 693     BDRVRBDState *s = bs->opaque;
 694     rbd_image_info_t info;
 695     int r;
 696
 697     r = rbd_stat(s->image, &info, sizeof(info));
 698     if (r < 0) {
 699         return r;
 700     }
 701
 702     bdi->cluster_size = info.obj_size;
 703     return 0;
 704 }
 705
 706 static int64_t qemu_rbd_getlength(BlockDriverState *bs)
 707 {
 708     BDRVRBDState *s = bs->opaque;
 709     rbd_image_info_t info;
 710     int r;
 711
 712     r = rbd_stat(s->image, &info, sizeof(info));
 713     if (r < 0) {
 714         return r;
 715     }
 716
 717     return info.size;
 718 }
 719
 720 static int qemu_rbd_truncate(BlockDriverState *bs, int64_t offset)
 721 {
 722     BDRVRBDState *s = bs->opaque;
 723     int r;
 724
 725     r = rbd_resize(s->image, offset);
 726     if (r < 0) {
 727         return r;
 728     }
 729
 730     return 0;
 731 }
 732
 733 static int qemu_rbd_snap_create(BlockDriverState *bs,
 734                                 QEMUSnapshotInfo *sn_info)
 735 {
 736     BDRVRBDState *s = bs->opaque;
 737     int r;
 738
 739     if (sn_info->name[0] == '\0') {
 740         return -EINVAL; /* we need a name for rbd snapshots */
 741     }
 742
 743     /*
 744      * rbd snapshots are using the name as the user controlled unique identifier
 745      * we can't use the rbd snapid for that purpose, as it can't be set
 746      */
 747     if (sn_info->id_str[0] != '\0' &&
 748         strcmp(sn_info->id_str, sn_info->name) != 0) {
 749         return -EINVAL;
 750     }
 751
 752     if (strlen(sn_info->name) >= sizeof(sn_info->id_str)) {
 753         return -ERANGE;
 754     }
 755
 756     r = rbd_snap_create(s->image, sn_info->name);
 757     if (r < 0) {
 758         error_report("failed to create snap: %s", strerror(-r));
 759         return r;
 760     }
 761
 762     return 0;
 763 }
 764
 765 static int qemu_rbd_snap_list(BlockDriverState *bs,
 766                               QEMUSnapshotInfo **psn_tab)
 767 {
 768     BDRVRBDState *s = bs->opaque;
 769     QEMUSnapshotInfo *sn_info, *sn_tab = NULL;
 770     int i, snap_count;
 771     rbd_snap_info_t *snaps;
 772     int max_snaps = RBD_MAX_SNAPS;
 773
 774     do {
 775         snaps = g_malloc(sizeof(*snaps) * max_snaps);
 776         snap_count = rbd_snap_list(s->image, snaps, &max_snaps);
 777         if (snap_count < 0) {
 778             g_free(snaps);
 779         }
 780     } while (snap_count == -ERANGE);
 781
 782     if (snap_count <= 0) {
 783         return snap_count;
 784     }
 785
 786     sn_tab = g_malloc0(snap_count * sizeof(QEMUSnapshotInfo));
 787
 788     for (i = 0; i < snap_count; i++) {
 789         const char *snap_name = snaps[i].name;
 790
 791         sn_info = sn_tab + i;
 792         pstrcpy(sn_info->id_str, sizeof(sn_info->id_str), snap_name);
 793         pstrcpy(sn_info->name, sizeof(sn_info->name), snap_name);
 794
 795         sn_info->vm_state_size = snaps[i].size;
 796         sn_info->date_sec = 0;
 797         sn_info->date_nsec = 0;
 798         sn_info->vm_clock_nsec = 0;
 799     }
 800     rbd_snap_list_end(snaps);
 801
 802     *psn_tab = sn_tab;
 803     return snap_count;
 804 }
 805
 806 static QEMUOptionParameter qemu_rbd_create_options[] = {
 807     {
 808      .name = BLOCK_OPT_SIZE,
 809      .type = OPT_SIZE,
 810      .help = "Virtual disk size"
 811     },
 812     {
 813      .name = BLOCK_OPT_CLUSTER_SIZE,
 814      .type = OPT_SIZE,
 815      .help = "RBD object size"
 816     },
 817     {NULL}
 818 };
 819
 820 static BlockDriver bdrv_rbd = {
 821     .format_name        = "rbd",
 822     .instance_size      = sizeof(BDRVRBDState),
 823     .bdrv_file_open     = qemu_rbd_open,
 824     .bdrv_close         = qemu_rbd_close,
 825     .bdrv_create        = qemu_rbd_create,
 826     .bdrv_get_info      = qemu_rbd_getinfo,
 827     .create_options     = qemu_rbd_create_options,
 828     .bdrv_getlength     = qemu_rbd_getlength,
 829     .bdrv_truncate      = qemu_rbd_truncate,
 830     .protocol_name      = "rbd",
 831
 832     .bdrv_aio_readv     = qemu_rbd_aio_readv,
 833     .bdrv_aio_writev    = qemu_rbd_aio_writev,
 834
 835     .bdrv_snapshot_create = qemu_rbd_snap_create,
 836     .bdrv_snapshot_list = qemu_rbd_snap_list,
 837 };
 838
 839 static void bdrv_rbd_init(void)
 840 {
 841     bdrv_register(&bdrv_rbd);
 842 }
 843
 844 block_init(bdrv_rbd_init);