block/rbd.c

   1 /*
   2  * QEMU Block driver for RADOS (Ceph)
   3  *
   4  * Copyright (C) 2010 Christian Brunner <chb@muc.de>
   5  *
   6  * This work is licensed under the terms of the GNU GPL, version 2.  See
   7  * the COPYING file in the top-level directory.
   8  *
   9  */
  10
  11 #include "qemu-common.h"
  12 #include "qemu-error.h"
  13
  14 #include "rbd_types.h"
  15 #include "block_int.h"
  16
  17 #include <rados/librados.h>
  18
  19
  20
  21 /*
  22  * When specifying the image filename use:
  23  *
  24  * rbd:poolname/devicename
  25  *
  26  * poolname must be the name of an existing rados pool
  27  *
  28  * devicename is the basename for all objects used to
  29  * emulate the raw device.
  30  *
  31  * Metadata information (image size, ...) is stored in an
  32  * object with the name "devicename.rbd".
  33  *
  34  * The raw device is split into 4MB sized objects by default.
  35  * The sequencenumber is encoded in a 12 byte long hex-string,
  36  * and is attached to the devicename, separated by a dot.
  37  * e.g. "devicename.1234567890ab"
  38  *
  39  */
  40
  41 #define OBJ_MAX_SIZE (1UL << OBJ_DEFAULT_OBJ_ORDER)
  42
  43 typedef struct RBDAIOCB {
  44     BlockDriverAIOCB common;
  45     QEMUBH *bh;
  46     int ret;
  47     QEMUIOVector *qiov;
  48     char *bounce;
  49     int write;
  50     int64_t sector_num;
  51     int aiocnt;
  52     int error;
  53     struct BDRVRBDState *s;
  54     int cancelled;
  55 } RBDAIOCB;
  56
  57 typedef struct RADOSCB {
  58     int rcbid;
  59     RBDAIOCB *acb;
  60     struct BDRVRBDState *s;
  61     int done;
  62     int64_t segsize;
  63     char *buf;
  64     int ret;
  65 } RADOSCB;
  66
  67 #define RBD_FD_READ 0
  68 #define RBD_FD_WRITE 1
  69
  70 typedef struct BDRVRBDState {
  71     int fds[2];
  72     rados_pool_t pool;
  73     rados_pool_t header_pool;
  74     char name[RBD_MAX_OBJ_NAME_SIZE];
  75     char block_name[RBD_MAX_BLOCK_NAME_SIZE];
  76     uint64_t size;
  77     uint64_t objsize;
  78     int qemu_aio_count;
  79     int event_reader_pos;
  80     RADOSCB *event_rcb;
  81 } BDRVRBDState;
  82
  83 typedef struct rbd_obj_header_ondisk RbdHeader1;
  84
  85 static void rbd_aio_bh_cb(void *opaque);
  86
  87 static int rbd_next_tok(char *dst, int dst_len,
  88                         char *src, char delim,
  89                         const char *name,
  90                         char **p)
  91 {
  92     int l;
  93     char *end;
  94
  95     *p = NULL;
  96
  97     if (delim != '\0') {
  98         end = strchr(src, delim);
  99         if (end) {
 100             *p = end + 1;
 101             *end = '\0';
 102         }
 103     }
 104     l = strlen(src);
 105     if (l >= dst_len) {
 106         error_report("%s too long", name);
 107         return -EINVAL;
 108     } else if (l == 0) {
 109         error_report("%s too short", name);
 110         return -EINVAL;
 111     }
 112
 113     pstrcpy(dst, dst_len, src);
 114
 115     return 0;
 116 }
 117
 118 static int rbd_parsename(const char *filename,
 119                          char *pool, int pool_len,
 120                          char *snap, int snap_len,
 121                          char *name, int name_len)
 122 {
 123     const char *start;
 124     char *p, *buf;
 125     int ret;
 126
 127     if (!strstart(filename, "rbd:", &start)) {
 128         return -EINVAL;
 129     }
 130
 131     buf = qemu_strdup(start);
 132     p = buf;
 133
 134     ret = rbd_next_tok(pool, pool_len, p, '/', "pool name", &p);
 135     if (ret < 0 || !p) {
 136         ret = -EINVAL;
 137         goto done;
 138     }
 139     ret = rbd_next_tok(name, name_len, p, '@', "object name", &p);
 140     if (ret < 0) {
 141         goto done;
 142     }
 143     if (!p) {
 144         *snap = '\0';
 145         goto done;
 146     }
 147
 148     ret = rbd_next_tok(snap, snap_len, p, '\0', "snap name", &p);
 149
 150 done:
 151     qemu_free(buf);
 152     return ret;
 153 }
 154
 155 static int create_tmap_op(uint8_t op, const char *name, char **tmap_desc)
 156 {
 157     uint32_t len = strlen(name);
 158     uint32_t len_le = cpu_to_le32(len);
 159     /* total_len = encoding op + name + empty buffer */
 160     uint32_t total_len = 1 + (sizeof(uint32_t) + len) + sizeof(uint32_t);
 161     uint8_t *desc = NULL;
 162
 163     desc = qemu_malloc(total_len);
 164
 165     *tmap_desc = (char *)desc;
 166
 167     *desc = op;
 168     desc++;
 169     memcpy(desc, &len_le, sizeof(len_le));
 170     desc += sizeof(len_le);
 171     memcpy(desc, name, len);
 172     desc += len;
 173     len = 0; /* no need for endian conversion for 0 */
 174     memcpy(desc, &len, sizeof(len));
 175     desc += sizeof(len);
 176
 177     return (char *)desc - *tmap_desc;
 178 }
 179
 180 static void free_tmap_op(char *tmap_desc)
 181 {
 182     qemu_free(tmap_desc);
 183 }
 184
 185 static int rbd_register_image(rados_pool_t pool, const char *name)
 186 {
 187     char *tmap_desc;
 188     const char *dir = RBD_DIRECTORY;
 189     int ret;
 190
 191     ret = create_tmap_op(CEPH_OSD_TMAP_SET, name, &tmap_desc);
 192     if (ret < 0) {
 193         return ret;
 194     }
 195
 196     ret = rados_tmap_update(pool, dir, tmap_desc, ret);
 197     free_tmap_op(tmap_desc);
 198
 199     return ret;
 200 }
 201
 202 static int touch_rbd_info(rados_pool_t pool, const char *info_oid)
 203 {
 204     int r = rados_write(pool, info_oid, 0, NULL, 0);
 205     if (r < 0) {
 206         return r;
 207     }
 208     return 0;
 209 }
 210
 211 static int rbd_assign_bid(rados_pool_t pool, uint64_t *id)
 212 {
 213     uint64_t out[1];
 214     const char *info_oid = RBD_INFO;
 215
 216     *id = 0;
 217
 218     int r = touch_rbd_info(pool, info_oid);
 219     if (r < 0) {
 220         return r;
 221     }
 222
 223     r = rados_exec(pool, info_oid, "rbd", "assign_bid", NULL,
 224                    0, (char *)out, sizeof(out));
 225     if (r < 0) {
 226         return r;
 227     }
 228
 229     le64_to_cpus(out);
 230     *id = out[0];
 231
 232     return 0;
 233 }
 234
 235 static int rbd_create(const char *filename, QEMUOptionParameter *options)
 236 {
 237     int64_t bytes = 0;
 238     int64_t objsize;
 239     uint64_t size;
 240     time_t mtime;
 241     uint8_t obj_order = RBD_DEFAULT_OBJ_ORDER;
 242     char pool[RBD_MAX_SEG_NAME_SIZE];
 243     char n[RBD_MAX_SEG_NAME_SIZE];
 244     char name[RBD_MAX_OBJ_NAME_SIZE];
 245     char snap_buf[RBD_MAX_SEG_NAME_SIZE];
 246     char *snap = NULL;
 247     RbdHeader1 header;
 248     rados_pool_t p;
 249     uint64_t bid;
 250     uint32_t hi, lo;
 251     int ret;
 252
 253     if (rbd_parsename(filename,
 254                       pool, sizeof(pool),
 255                       snap_buf, sizeof(snap_buf),
 256                       name, sizeof(name)) < 0) {
 257         return -EINVAL;
 258     }
 259     if (snap_buf[0] != '\0') {
 260         snap = snap_buf;
 261     }
 262
 263     snprintf(n, sizeof(n), "%s%s", name, RBD_SUFFIX);
 264
 265     /* Read out options */
 266     while (options && options->name) {
 267         if (!strcmp(options->name, BLOCK_OPT_SIZE)) {
 268             bytes = options->value.n;
 269         } else if (!strcmp(options->name, BLOCK_OPT_CLUSTER_SIZE)) {
 270             if (options->value.n) {
 271                 objsize = options->value.n;
 272                 if ((objsize - 1) & objsize) {    /* not a power of 2? */
 273                     error_report("obj size needs to be power of 2");
 274                     return -EINVAL;
 275                 }
 276                 if (objsize < 4096) {
 277                     error_report("obj size too small");
 278                     return -EINVAL;
 279                 }
 280                 obj_order = ffs(objsize) - 1;
 281             }
 282         }
 283         options++;
 284     }
 285
 286     memset(&header, 0, sizeof(header));
 287     pstrcpy(header.text, sizeof(header.text), RBD_HEADER_TEXT);
 288     pstrcpy(header.signature, sizeof(header.signature), RBD_HEADER_SIGNATURE);
 289     pstrcpy(header.version, sizeof(header.version), RBD_HEADER_VERSION);
 290     header.image_size = cpu_to_le64(bytes);
 291     header.options.order = obj_order;
 292     header.options.crypt_type = RBD_CRYPT_NONE;
 293     header.options.comp_type = RBD_COMP_NONE;
 294     header.snap_seq = 0;
 295     header.snap_count = 0;
 296
 297     if (rados_initialize(0, NULL) < 0) {
 298         error_report("error initializing");
 299         return -EIO;
 300     }
 301
 302     if (rados_open_pool(pool, &p)) {
 303         error_report("error opening pool %s", pool);
 304         rados_deinitialize();
 305         return -EIO;
 306     }
 307
 308     /* check for existing rbd header file */
 309     ret = rados_stat(p, n, &size, &mtime);
 310     if (ret == 0) {
 311         ret=-EEXIST;
 312         goto done;
 313     }
 314
 315     ret = rbd_assign_bid(p, &bid);
 316     if (ret < 0) {
 317         error_report("failed assigning block id");
 318         rados_deinitialize();
 319         return -EIO;
 320     }
 321     hi = bid >> 32;
 322     lo = bid & 0xFFFFFFFF;
 323     snprintf(header.block_name, sizeof(header.block_name), "rb.%x.%x", hi, lo);
 324
 325     /* create header file */
 326     ret = rados_write(p, n, 0, (const char *)&header, sizeof(header));
 327     if (ret < 0) {
 328         goto done;
 329     }
 330
 331     ret = rbd_register_image(p, name);
 332 done:
 333     rados_close_pool(p);
 334     rados_deinitialize();
 335
 336     return ret;
 337 }
 338
 339 /*
 340  * This aio completion is being called from rbd_aio_event_reader() and
 341  * runs in qemu context. It schedules a bh, but just in case the aio
 342  * was not cancelled before.
 343  */
 344 static void rbd_complete_aio(RADOSCB *rcb)
 345 {
 346     RBDAIOCB *acb = rcb->acb;
 347     int64_t r;
 348
 349     acb->aiocnt--;
 350
 351     if (acb->cancelled) {
 352         if (!acb->aiocnt) {
 353             qemu_vfree(acb->bounce);
 354             qemu_aio_release(acb);
 355         }
 356         goto done;
 357     }
 358
 359     r = rcb->ret;
 360
 361     if (acb->write) {
 362         if (r < 0) {
 363             acb->ret = r;
 364             acb->error = 1;
 365         } else if (!acb->error) {
 366             acb->ret += rcb->segsize;
 367         }
 368     } else {
 369         if (r == -ENOENT) {
 370             memset(rcb->buf, 0, rcb->segsize);
 371             if (!acb->error) {
 372                 acb->ret += rcb->segsize;
 373             }
 374         } else if (r < 0) {
 375             memset(rcb->buf, 0, rcb->segsize);
 376             acb->ret = r;
 377             acb->error = 1;
 378         } else if (r < rcb->segsize) {
 379             memset(rcb->buf + r, 0, rcb->segsize - r);
 380             if (!acb->error) {
 381                 acb->ret += rcb->segsize;
 382             }
 383         } else if (!acb->error) {
 384             acb->ret += r;
 385         }
 386     }
 387     /* Note that acb->bh can be NULL in case where the aio was cancelled */
 388     if (!acb->aiocnt) {
 389         acb->bh = qemu_bh_new(rbd_aio_bh_cb, acb);
 390         qemu_bh_schedule(acb->bh);
 391     }
 392 done:
 393     qemu_free(rcb);
 394 }
 395
 396 /*
 397  * aio fd read handler. It runs in the qemu context and calls the
 398  * completion handling of completed rados aio operations.
 399  */
 400 static void rbd_aio_event_reader(void *opaque)
 401 {
 402     BDRVRBDState *s = opaque;
 403
 404     ssize_t ret;
 405
 406     do {
 407         char *p = (char *)&s->event_rcb;
 408
 409         /* now read the rcb pointer that was sent from a non qemu thread */
 410         if ((ret = read(s->fds[RBD_FD_READ], p + s->event_reader_pos,
 411                         sizeof(s->event_rcb) - s->event_reader_pos)) > 0) {
 412             if (ret > 0) {
 413                 s->event_reader_pos += ret;
 414                 if (s->event_reader_pos == sizeof(s->event_rcb)) {
 415                     s->event_reader_pos = 0;
 416                     rbd_complete_aio(s->event_rcb);
 417                     s->qemu_aio_count --;
 418                 }
 419             }
 420         }
 421     } while (ret < 0 && errno == EINTR);
 422 }
 423
 424 static int rbd_aio_flush_cb(void *opaque)
 425 {
 426     BDRVRBDState *s = opaque;
 427
 428     return (s->qemu_aio_count > 0);
 429 }
 430
 431
 432 static int rbd_set_snapc(rados_pool_t pool, const char *snap, RbdHeader1 *header)
 433 {
 434     uint32_t snap_count = le32_to_cpu(header->snap_count);
 435     rados_snap_t *snaps = NULL;
 436     rados_snap_t seq;
 437     uint32_t i;
 438     uint64_t snap_names_len = le64_to_cpu(header->snap_names_len);
 439     int r;
 440     rados_snap_t snapid = 0;
 441
 442     if (snap_count) {
 443         const char *header_snap = (const char *)&header->snaps[snap_count];
 444         const char *end = header_snap + snap_names_len;
 445         snaps = qemu_malloc(sizeof(rados_snap_t) * header->snap_count);
 446
 447         for (i=0; i < snap_count; i++) {
 448             snaps[i] = le64_to_cpu(header->snaps[i].id);
 449
 450             if (snap && strcmp(snap, header_snap) == 0) {
 451                 snapid = snaps[i];
 452             }
 453
 454             header_snap += strlen(header_snap) + 1;
 455             if (header_snap > end) {
 456                 error_report("bad header, snapshot list broken");
 457             }
 458         }
 459     }
 460
 461     if (snap && !snapid) {
 462         error_report("snapshot not found");
 463         qemu_free(snaps);
 464         return -ENOENT;
 465     }
 466     seq = le32_to_cpu(header->snap_seq);
 467
 468     r = rados_set_snap_context(pool, seq, snaps, snap_count);
 469
 470     rados_set_snap(pool, snapid);
 471
 472     qemu_free(snaps);
 473
 474     return r;
 475 }
 476
 477 #define BUF_READ_START_LEN    4096
 478
 479 static int rbd_read_header(BDRVRBDState *s, char **hbuf)
 480 {
 481     char *buf = NULL;
 482     char n[RBD_MAX_SEG_NAME_SIZE];
 483     uint64_t len = BUF_READ_START_LEN;
 484     int r;
 485
 486     snprintf(n, sizeof(n), "%s%s", s->name, RBD_SUFFIX);
 487
 488     buf = qemu_malloc(len);
 489
 490     r = rados_read(s->header_pool, n, 0, buf, len);
 491     if (r < 0) {
 492         goto failed;
 493     }
 494
 495     if (r < len) {
 496         goto done;
 497     }
 498
 499     qemu_free(buf);
 500     buf = qemu_malloc(len);
 501
 502     r = rados_stat(s->header_pool, n, &len, NULL);
 503     if (r < 0) {
 504         goto failed;
 505     }
 506
 507     r = rados_read(s->header_pool, n, 0, buf, len);
 508     if (r < 0) {
 509         goto failed;
 510     }
 511
 512 done:
 513     *hbuf = buf;
 514     return 0;
 515
 516 failed:
 517     qemu_free(buf);
 518     return r;
 519 }
 520
 521 static int rbd_open(BlockDriverState *bs, const char *filename, int flags)
 522 {
 523     BDRVRBDState *s = bs->opaque;
 524     RbdHeader1 *header;
 525     char pool[RBD_MAX_SEG_NAME_SIZE];
 526     char snap_buf[RBD_MAX_SEG_NAME_SIZE];
 527     char *snap = NULL;
 528     char *hbuf = NULL;
 529     int r;
 530
 531     if (rbd_parsename(filename, pool, sizeof(pool),
 532                       snap_buf, sizeof(snap_buf),
 533                       s->name, sizeof(s->name)) < 0) {
 534         return -EINVAL;
 535     }
 536     if (snap_buf[0] != '\0') {
 537         snap = snap_buf;
 538     }
 539
 540     if ((r = rados_initialize(0, NULL)) < 0) {
 541         error_report("error initializing");
 542         return r;
 543     }
 544
 545     if ((r = rados_open_pool(pool, &s->pool))) {
 546         error_report("error opening pool %s", pool);
 547         rados_deinitialize();
 548         return r;
 549     }
 550
 551     if ((r = rados_open_pool(pool, &s->header_pool))) {
 552         error_report("error opening pool %s", pool);
 553         rados_deinitialize();
 554         return r;
 555     }
 556
 557     if ((r = rbd_read_header(s, &hbuf)) < 0) {
 558         error_report("error reading header from %s", s->name);
 559         goto failed;
 560     }
 561
 562     if (memcmp(hbuf + 64, RBD_HEADER_SIGNATURE, 4)) {
 563         error_report("Invalid header signature");
 564         r = -EMEDIUMTYPE;
 565         goto failed;
 566     }
 567
 568     if (memcmp(hbuf + 68, RBD_HEADER_VERSION, 8)) {
 569         error_report("Unknown image version");
 570         r = -EMEDIUMTYPE;
 571         goto failed;
 572     }
 573
 574     header = (RbdHeader1 *) hbuf;
 575     s->size = le64_to_cpu(header->image_size);
 576     s->objsize = 1ULL << header->options.order;
 577     memcpy(s->block_name, header->block_name, sizeof(header->block_name));
 578
 579     r = rbd_set_snapc(s->pool, snap, header);
 580     if (r < 0) {
 581         error_report("failed setting snap context: %s", strerror(-r));
 582         goto failed;
 583     }
 584
 585     bs->read_only = (snap != NULL);
 586
 587     s->event_reader_pos = 0;
 588     r = qemu_pipe(s->fds);
 589     if (r < 0) {
 590         error_report("error opening eventfd");
 591         goto failed;
 592     }
 593     fcntl(s->fds[0], F_SETFL, O_NONBLOCK);
 594     fcntl(s->fds[1], F_SETFL, O_NONBLOCK);
 595     qemu_aio_set_fd_handler(s->fds[RBD_FD_READ], rbd_aio_event_reader, NULL,
 596         rbd_aio_flush_cb, NULL, s);
 597
 598     qemu_free(hbuf);
 599
 600     return 0;
 601
 602 failed:
 603     qemu_free(hbuf);
 604
 605     rados_close_pool(s->header_pool);
 606     rados_close_pool(s->pool);
 607     rados_deinitialize();
 608     return r;
 609 }
 610
 611 static void rbd_close(BlockDriverState *bs)
 612 {
 613     BDRVRBDState *s = bs->opaque;
 614
 615     close(s->fds[0]);
 616     close(s->fds[1]);
 617     qemu_aio_set_fd_handler(s->fds[RBD_FD_READ], NULL , NULL, NULL, NULL,
 618         NULL);
 619
 620     rados_close_pool(s->header_pool);
 621     rados_close_pool(s->pool);
 622     rados_deinitialize();
 623 }
 624
 625 /*
 626  * Cancel aio. Since we don't reference acb in a non qemu threads,
 627  * it is safe to access it here.
 628  */
 629 static void rbd_aio_cancel(BlockDriverAIOCB *blockacb)
 630 {
 631     RBDAIOCB *acb = (RBDAIOCB *) blockacb;
 632     acb->cancelled = 1;
 633 }
 634
 635 static AIOPool rbd_aio_pool = {
 636     .aiocb_size = sizeof(RBDAIOCB),
 637     .cancel = rbd_aio_cancel,
 638 };
 639
 640 /*
 641  * This is the callback function for rados_aio_read and _write
 642  *
 643  * Note: this function is being called from a non qemu thread so
 644  * we need to be careful about what we do here. Generally we only
 645  * write to the block notification pipe, and do the rest of the
 646  * io completion handling from rbd_aio_event_reader() which
 647  * runs in a qemu context.
 648  */
 649 static void rbd_finish_aiocb(rados_completion_t c, RADOSCB *rcb)
 650 {
 651     int ret;
 652     rcb->ret = rados_aio_get_return_value(c);
 653     rados_aio_release(c);
 654     while (1) {
 655         fd_set wfd;
 656         int fd = rcb->s->fds[RBD_FD_WRITE];
 657
 658         /* send the rcb pointer to the qemu thread that is responsible
 659            for the aio completion. Must do it in a qemu thread context */
 660         ret = write(fd, (void *)&rcb, sizeof(rcb));
 661         if (ret >= 0) {
 662             break;
 663         }
 664         if (errno == EINTR) {
 665             continue;
 666         }
 667         if (errno != EAGAIN) {
 668             break;
 669         }
 670
 671         FD_ZERO(&wfd);
 672         FD_SET(fd, &wfd);
 673         do {
 674             ret = select(fd + 1, NULL, &wfd, NULL, NULL);
 675         } while (ret < 0 && errno == EINTR);
 676     }
 677
 678     if (ret < 0) {
 679         error_report("failed writing to acb->s->fds\n");
 680         qemu_free(rcb);
 681     }
 682 }
 683
 684 /* Callback when all queued rados_aio requests are complete */
 685
 686 static void rbd_aio_bh_cb(void *opaque)
 687 {
 688     RBDAIOCB *acb = opaque;
 689
 690     if (!acb->write) {
 691         qemu_iovec_from_buffer(acb->qiov, acb->bounce, acb->qiov->size);
 692     }
 693     qemu_vfree(acb->bounce);
 694     acb->common.cb(acb->common.opaque, (acb->ret > 0 ? 0 : acb->ret));
 695     qemu_bh_delete(acb->bh);
 696     acb->bh = NULL;
 697
 698     qemu_aio_release(acb);
 699 }
 700
 701 static BlockDriverAIOCB *rbd_aio_rw_vector(BlockDriverState *bs,
 702                                            int64_t sector_num,
 703                                            QEMUIOVector *qiov,
 704                                            int nb_sectors,
 705                                            BlockDriverCompletionFunc *cb,
 706                                            void *opaque, int write)
 707 {
 708     RBDAIOCB *acb;
 709     RADOSCB *rcb;
 710     rados_completion_t c;
 711     char n[RBD_MAX_SEG_NAME_SIZE];
 712     int64_t segnr, segoffs, segsize, last_segnr;
 713     int64_t off, size;
 714     char *buf;
 715
 716     BDRVRBDState *s = bs->opaque;
 717
 718     acb = qemu_aio_get(&rbd_aio_pool, bs, cb, opaque);
 719     acb->write = write;
 720     acb->qiov = qiov;
 721     acb->bounce = qemu_blockalign(bs, qiov->size);
 722     acb->aiocnt = 0;
 723     acb->ret = 0;
 724     acb->error = 0;
 725     acb->s = s;
 726     acb->cancelled = 0;
 727     acb->bh = NULL;
 728
 729     if (write) {
 730         qemu_iovec_to_buffer(acb->qiov, acb->bounce);
 731     }
 732
 733     buf = acb->bounce;
 734
 735     off = sector_num * BDRV_SECTOR_SIZE;
 736     size = nb_sectors * BDRV_SECTOR_SIZE;
 737     segnr = off / s->objsize;
 738     segoffs = off % s->objsize;
 739     segsize = s->objsize - segoffs;
 740
 741     last_segnr = ((off + size - 1) / s->objsize);
 742     acb->aiocnt = (last_segnr - segnr) + 1;
 743
 744     s->qemu_aio_count += acb->aiocnt; /* All the RADOSCB */
 745
 746     while (size > 0) {
 747         if (size < segsize) {
 748             segsize = size;
 749         }
 750
 751         snprintf(n, sizeof(n), "%s.%012" PRIx64, s->block_name,
 752                  segnr);
 753
 754         rcb = qemu_malloc(sizeof(RADOSCB));
 755         rcb->done = 0;
 756         rcb->acb = acb;
 757         rcb->segsize = segsize;
 758         rcb->buf = buf;
 759         rcb->s = acb->s;
 760
 761         if (write) {
 762             rados_aio_create_completion(rcb, NULL,
 763                                         (rados_callback_t) rbd_finish_aiocb,
 764                                         &c);
 765             rados_aio_write(s->pool, n, segoffs, buf, segsize, c);
 766         } else {
 767             rados_aio_create_completion(rcb,
 768                                         (rados_callback_t) rbd_finish_aiocb,
 769                                         NULL, &c);
 770             rados_aio_read(s->pool, n, segoffs, buf, segsize, c);
 771         }
 772
 773         buf += segsize;
 774         size -= segsize;
 775         segoffs = 0;
 776         segsize = s->objsize;
 777         segnr++;
 778     }
 779
 780     return &acb->common;
 781 }
 782
 783 static BlockDriverAIOCB *rbd_aio_readv(BlockDriverState * bs,
 784                                        int64_t sector_num, QEMUIOVector * qiov,
 785                                        int nb_sectors,
 786                                        BlockDriverCompletionFunc * cb,
 787                                        void *opaque)
 788 {
 789     return rbd_aio_rw_vector(bs, sector_num, qiov, nb_sectors, cb, opaque, 0);
 790 }
 791
 792 static BlockDriverAIOCB *rbd_aio_writev(BlockDriverState * bs,
 793                                         int64_t sector_num, QEMUIOVector * qiov,
 794                                         int nb_sectors,
 795                                         BlockDriverCompletionFunc * cb,
 796                                         void *opaque)
 797 {
 798     return rbd_aio_rw_vector(bs, sector_num, qiov, nb_sectors, cb, opaque, 1);
 799 }
 800
 801 static int rbd_getinfo(BlockDriverState * bs, BlockDriverInfo * bdi)
 802 {
 803     BDRVRBDState *s = bs->opaque;
 804     bdi->cluster_size = s->objsize;
 805     return 0;
 806 }
 807
 808 static int64_t rbd_getlength(BlockDriverState * bs)
 809 {
 810     BDRVRBDState *s = bs->opaque;
 811
 812     return s->size;
 813 }
 814
 815 static int rbd_snap_create(BlockDriverState *bs, QEMUSnapshotInfo *sn_info)
 816 {
 817     BDRVRBDState *s = bs->opaque;
 818     char inbuf[512], outbuf[128];
 819     uint64_t snap_id;
 820     int r;
 821     char *p = inbuf;
 822     char *end = inbuf + sizeof(inbuf);
 823     char n[RBD_MAX_SEG_NAME_SIZE];
 824     char *hbuf = NULL;
 825     RbdHeader1 *header;
 826
 827     if (sn_info->name[0] == '\0') {
 828         return -EINVAL; /* we need a name for rbd snapshots */
 829     }
 830
 831     /*
 832      * rbd snapshots are using the name as the user controlled unique identifier
 833      * we can't use the rbd snapid for that purpose, as it can't be set
 834      */
 835     if (sn_info->id_str[0] != '\0' &&
 836         strcmp(sn_info->id_str, sn_info->name) != 0) {
 837         return -EINVAL;
 838     }
 839
 840     if (strlen(sn_info->name) >= sizeof(sn_info->id_str)) {
 841         return -ERANGE;
 842     }
 843
 844     r = rados_selfmanaged_snap_create(s->header_pool, &snap_id);
 845     if (r < 0) {
 846         error_report("failed to create snap id: %s", strerror(-r));
 847         return r;
 848     }
 849
 850     *(uint32_t *)p = strlen(sn_info->name);
 851     cpu_to_le32s((uint32_t *)p);
 852     p += sizeof(uint32_t);
 853     strncpy(p, sn_info->name, end - p);
 854     p += strlen(p);
 855     if (p + sizeof(snap_id) > end) {
 856         error_report("invalid input parameter");
 857         return -EINVAL;
 858     }
 859
 860     *(uint64_t *)p = snap_id;
 861     cpu_to_le64s((uint64_t *)p);
 862
 863     snprintf(n, sizeof(n), "%s%s", s->name, RBD_SUFFIX);
 864
 865     r = rados_exec(s->header_pool, n, "rbd", "snap_add", inbuf,
 866                    sizeof(inbuf), outbuf, sizeof(outbuf));
 867     if (r < 0) {
 868         error_report("rbd.snap_add execution failed failed: %s", strerror(-r));
 869         return r;
 870     }
 871
 872     sprintf(sn_info->id_str, "%s", sn_info->name);
 873
 874     r = rbd_read_header(s, &hbuf);
 875     if (r < 0) {
 876         error_report("failed reading header: %s", strerror(-r));
 877         return r;
 878     }
 879
 880     header = (RbdHeader1 *) hbuf;
 881     r = rbd_set_snapc(s->pool, sn_info->name, header);
 882     if (r < 0) {
 883         error_report("failed setting snap context: %s", strerror(-r));
 884         goto failed;
 885     }
 886
 887     return 0;
 888
 889 failed:
 890     qemu_free(header);
 891     return r;
 892 }
 893
 894 static int decode32(char **p, const char *end, uint32_t *v)
 895 {
 896     if (*p + 4 > end) {
 897         return -ERANGE;
 898     }
 899
 900     *v = *(uint32_t *)(*p);
 901     le32_to_cpus(v);
 902     *p += 4;
 903     return 0;
 904 }
 905
 906 static int decode64(char **p, const char *end, uint64_t *v)
 907 {
 908     if (*p + 8 > end) {
 909         return -ERANGE;
 910     }
 911
 912     *v = *(uint64_t *)(*p);
 913     le64_to_cpus(v);
 914     *p += 8;
 915     return 0;
 916 }
 917
 918 static int decode_str(char **p, const char *end, char **s)
 919 {
 920     uint32_t len;
 921     int r;
 922
 923     if ((r = decode32(p, end, &len)) < 0) {
 924         return r;
 925     }
 926
 927     *s = qemu_malloc(len + 1);
 928     memcpy(*s, *p, len);
 929     *p += len;
 930     (*s)[len] = '\0';
 931
 932     return len;
 933 }
 934
 935 static int rbd_snap_list(BlockDriverState *bs, QEMUSnapshotInfo **psn_tab)
 936 {
 937     BDRVRBDState *s = bs->opaque;
 938     char n[RBD_MAX_SEG_NAME_SIZE];
 939     QEMUSnapshotInfo *sn_info, *sn_tab = NULL;
 940     RbdHeader1 *header;
 941     char *hbuf = NULL;
 942     char *outbuf = NULL, *end, *buf;
 943     uint64_t len;
 944     uint64_t snap_seq;
 945     uint32_t snap_count;
 946     int r, i;
 947
 948     /* read header to estimate how much space we need to read the snap
 949      * list */
 950     if ((r = rbd_read_header(s, &hbuf)) < 0) {
 951         goto done_err;
 952     }
 953     header = (RbdHeader1 *)hbuf;
 954     len = le64_to_cpu(header->snap_names_len);
 955     len += 1024; /* should have already been enough, but new snapshots might
 956                     already been created since we read the header. just allocate
 957                     a bit more, so that in most cases it'll suffice anyway */
 958     qemu_free(hbuf);
 959
 960     snprintf(n, sizeof(n), "%s%s", s->name, RBD_SUFFIX);
 961     while (1) {
 962         qemu_free(outbuf);
 963         outbuf = qemu_malloc(len);
 964
 965         r = rados_exec(s->header_pool, n, "rbd", "snap_list", NULL, 0,
 966                        outbuf, len);
 967         if (r < 0) {
 968             error_report("rbd.snap_list execution failed failed: %s", strerror(-r));
 969             goto done_err;
 970         }
 971         if (r != len) {
 972             break;
 973         }
 974
 975         /* if we're here, we probably raced with some snaps creation */
 976         len *= 2;
 977     }
 978     buf = outbuf;
 979     end = buf + len;
 980
 981     if ((r = decode64(&buf, end, &snap_seq)) < 0) {
 982         goto done_err;
 983     }
 984     if ((r = decode32(&buf, end, &snap_count)) < 0) {
 985         goto done_err;
 986     }
 987
 988     sn_tab = qemu_mallocz(snap_count * sizeof(QEMUSnapshotInfo));
 989     for (i = 0; i < snap_count; i++) {
 990         uint64_t id, image_size;
 991         char *snap_name;
 992
 993         if ((r = decode64(&buf, end, &id)) < 0) {
 994             goto done_err;
 995         }
 996         if ((r = decode64(&buf, end, &image_size)) < 0) {
 997             goto done_err;
 998         }
 999         if ((r = decode_str(&buf, end, &snap_name)) < 0) {
1000             goto done_err;
1001         }
1002
1003         sn_info = sn_tab + i;
1004         pstrcpy(sn_info->id_str, sizeof(sn_info->id_str), snap_name);
1005         pstrcpy(sn_info->name, sizeof(sn_info->name), snap_name);
1006         qemu_free(snap_name);
1007
1008         sn_info->vm_state_size = image_size;
1009         sn_info->date_sec = 0;
1010         sn_info->date_nsec = 0;
1011         sn_info->vm_clock_nsec = 0;
1012     }
1013     *psn_tab = sn_tab;
1014     qemu_free(outbuf);
1015     return snap_count;
1016 done_err:
1017     qemu_free(sn_tab);
1018     qemu_free(outbuf);
1019     return r;
1020 }
1021
1022 static QEMUOptionParameter rbd_create_options[] = {
1023     {
1024      .name = BLOCK_OPT_SIZE,
1025      .type = OPT_SIZE,
1026      .help = "Virtual disk size"
1027     },
1028     {
1029      .name = BLOCK_OPT_CLUSTER_SIZE,
1030      .type = OPT_SIZE,
1031      .help = "RBD object size"
1032     },
1033     {NULL}
1034 };
1035
1036 static BlockDriver bdrv_rbd = {
1037     .format_name        = "rbd",
1038     .instance_size      = sizeof(BDRVRBDState),
1039     .bdrv_file_open     = rbd_open,
1040     .bdrv_close         = rbd_close,
1041     .bdrv_create        = rbd_create,
1042     .bdrv_get_info      = rbd_getinfo,
1043     .create_options     = rbd_create_options,
1044     .bdrv_getlength     = rbd_getlength,
1045     .protocol_name      = "rbd",
1046
1047     .bdrv_aio_readv     = rbd_aio_readv,
1048     .bdrv_aio_writev    = rbd_aio_writev,
1049
1050     .bdrv_snapshot_create = rbd_snap_create,
1051     .bdrv_snapshot_list = rbd_snap_list,
1052 };
1053
1054 static void bdrv_rbd_init(void)
1055 {
1056     bdrv_register(&bdrv_rbd);
1057 }
1058
1059 block_init(bdrv_rbd_init);