qemu-char: Avoid unused variable warning in some configs
[qemu/pbrook.git] / block / rbd.c
blob8cd10a7b59421cb778c9ac57a009f466f16688b9
1 /*
2 * QEMU Block driver for RADOS (Ceph)
4 * Copyright (C) 2010-2011 Christian Brunner <chb@muc.de>,
5 * Josh Durgin <josh.durgin@dreamhost.com>
7 * This work is licensed under the terms of the GNU GPL, version 2. See
8 * the COPYING file in the top-level directory.
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.
14 #include <inttypes.h>
16 #include "qemu-common.h"
17 #include "qemu/error-report.h"
18 #include "block/block_int.h"
20 #include <rbd/librbd.h>
23 * When specifying the image filename use:
25 * rbd:poolname/devicename[@snapshotname][:option1=value1[:option2=value2...]]
27 * poolname must be the name of an existing rados pool.
29 * devicename is the name of the rbd image.
31 * Each option given is used to configure rados, and may be any valid
32 * Ceph option, "id", or "conf".
34 * The "id" option indicates what user we should authenticate as to
35 * the Ceph cluster. If it is excluded we will use the Ceph default
36 * (normally 'admin').
38 * The "conf" option specifies a Ceph configuration file to read. If
39 * it is not specified, we will read from the default Ceph locations
40 * (e.g., /etc/ceph/ceph.conf). To avoid reading _any_ configuration
41 * file, specify conf=/dev/null.
43 * Configuration values containing :, @, or = can be escaped with a
44 * leading "\".
47 /* rbd_aio_discard added in 0.1.2 */
48 #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 2)
49 #define LIBRBD_SUPPORTS_DISCARD
50 #else
51 #undef LIBRBD_SUPPORTS_DISCARD
52 #endif
54 #define OBJ_MAX_SIZE (1UL << OBJ_DEFAULT_OBJ_ORDER)
56 #define RBD_MAX_CONF_NAME_SIZE 128
57 #define RBD_MAX_CONF_VAL_SIZE 512
58 #define RBD_MAX_CONF_SIZE 1024
59 #define RBD_MAX_POOL_NAME_SIZE 128
60 #define RBD_MAX_SNAP_NAME_SIZE 128
61 #define RBD_MAX_SNAPS 100
63 typedef enum {
64 RBD_AIO_READ,
65 RBD_AIO_WRITE,
66 RBD_AIO_DISCARD
67 } RBDAIOCmd;
69 typedef struct RBDAIOCB {
70 BlockDriverAIOCB common;
71 QEMUBH *bh;
72 int64_t ret;
73 QEMUIOVector *qiov;
74 char *bounce;
75 RBDAIOCmd cmd;
76 int64_t sector_num;
77 int error;
78 struct BDRVRBDState *s;
79 int cancelled;
80 int status;
81 } RBDAIOCB;
83 typedef struct RADOSCB {
84 int rcbid;
85 RBDAIOCB *acb;
86 struct BDRVRBDState *s;
87 int done;
88 int64_t size;
89 char *buf;
90 int64_t ret;
91 } RADOSCB;
93 #define RBD_FD_READ 0
94 #define RBD_FD_WRITE 1
96 typedef struct BDRVRBDState {
97 int fds[2];
98 rados_t cluster;
99 rados_ioctx_t io_ctx;
100 rbd_image_t image;
101 char name[RBD_MAX_IMAGE_NAME_SIZE];
102 int qemu_aio_count;
103 char *snap;
104 int event_reader_pos;
105 RADOSCB *event_rcb;
106 } BDRVRBDState;
108 static void rbd_aio_bh_cb(void *opaque);
110 static int qemu_rbd_next_tok(char *dst, int dst_len,
111 char *src, char delim,
112 const char *name,
113 char **p)
115 int l;
116 char *end;
118 *p = NULL;
120 if (delim != '\0') {
121 for (end = src; *end; ++end) {
122 if (*end == delim) {
123 break;
125 if (*end == '\\' && end[1] != '\0') {
126 end++;
129 if (*end == delim) {
130 *p = end + 1;
131 *end = '\0';
134 l = strlen(src);
135 if (l >= dst_len) {
136 error_report("%s too long", name);
137 return -EINVAL;
138 } else if (l == 0) {
139 error_report("%s too short", name);
140 return -EINVAL;
143 pstrcpy(dst, dst_len, src);
145 return 0;
148 static void qemu_rbd_unescape(char *src)
150 char *p;
152 for (p = src; *src; ++src, ++p) {
153 if (*src == '\\' && src[1] != '\0') {
154 src++;
156 *p = *src;
158 *p = '\0';
161 static int qemu_rbd_parsename(const char *filename,
162 char *pool, int pool_len,
163 char *snap, int snap_len,
164 char *name, int name_len,
165 char *conf, int conf_len)
167 const char *start;
168 char *p, *buf;
169 int ret;
171 if (!strstart(filename, "rbd:", &start)) {
172 return -EINVAL;
175 buf = g_strdup(start);
176 p = buf;
177 *snap = '\0';
178 *conf = '\0';
180 ret = qemu_rbd_next_tok(pool, pool_len, p, '/', "pool name", &p);
181 if (ret < 0 || !p) {
182 ret = -EINVAL;
183 goto done;
185 qemu_rbd_unescape(pool);
187 if (strchr(p, '@')) {
188 ret = qemu_rbd_next_tok(name, name_len, p, '@', "object name", &p);
189 if (ret < 0) {
190 goto done;
192 ret = qemu_rbd_next_tok(snap, snap_len, p, ':', "snap name", &p);
193 qemu_rbd_unescape(snap);
194 } else {
195 ret = qemu_rbd_next_tok(name, name_len, p, ':', "object name", &p);
197 qemu_rbd_unescape(name);
198 if (ret < 0 || !p) {
199 goto done;
202 ret = qemu_rbd_next_tok(conf, conf_len, p, '\0', "configuration", &p);
204 done:
205 g_free(buf);
206 return ret;
209 static char *qemu_rbd_parse_clientname(const char *conf, char *clientname)
211 const char *p = conf;
213 while (*p) {
214 int len;
215 const char *end = strchr(p, ':');
217 if (end) {
218 len = end - p;
219 } else {
220 len = strlen(p);
223 if (strncmp(p, "id=", 3) == 0) {
224 len -= 3;
225 strncpy(clientname, p + 3, len);
226 clientname[len] = '\0';
227 return clientname;
229 if (end == NULL) {
230 break;
232 p = end + 1;
234 return NULL;
237 static int qemu_rbd_set_conf(rados_t cluster, const char *conf)
239 char *p, *buf;
240 char name[RBD_MAX_CONF_NAME_SIZE];
241 char value[RBD_MAX_CONF_VAL_SIZE];
242 int ret = 0;
244 buf = g_strdup(conf);
245 p = buf;
247 while (p) {
248 ret = qemu_rbd_next_tok(name, sizeof(name), p,
249 '=', "conf option name", &p);
250 if (ret < 0) {
251 break;
253 qemu_rbd_unescape(name);
255 if (!p) {
256 error_report("conf option %s has no value", name);
257 ret = -EINVAL;
258 break;
261 ret = qemu_rbd_next_tok(value, sizeof(value), p,
262 ':', "conf option value", &p);
263 if (ret < 0) {
264 break;
266 qemu_rbd_unescape(value);
268 if (strcmp(name, "conf") == 0) {
269 ret = rados_conf_read_file(cluster, value);
270 if (ret < 0) {
271 error_report("error reading conf file %s", value);
272 break;
274 } else if (strcmp(name, "id") == 0) {
275 /* ignore, this is parsed by qemu_rbd_parse_clientname() */
276 } else {
277 ret = rados_conf_set(cluster, name, value);
278 if (ret < 0) {
279 error_report("invalid conf option %s", name);
280 ret = -EINVAL;
281 break;
286 g_free(buf);
287 return ret;
290 static int qemu_rbd_create(const char *filename, QEMUOptionParameter *options)
292 int64_t bytes = 0;
293 int64_t objsize;
294 int obj_order = 0;
295 char pool[RBD_MAX_POOL_NAME_SIZE];
296 char name[RBD_MAX_IMAGE_NAME_SIZE];
297 char snap_buf[RBD_MAX_SNAP_NAME_SIZE];
298 char conf[RBD_MAX_CONF_SIZE];
299 char clientname_buf[RBD_MAX_CONF_SIZE];
300 char *clientname;
301 rados_t cluster;
302 rados_ioctx_t io_ctx;
303 int ret;
305 if (qemu_rbd_parsename(filename, pool, sizeof(pool),
306 snap_buf, sizeof(snap_buf),
307 name, sizeof(name),
308 conf, sizeof(conf)) < 0) {
309 return -EINVAL;
312 /* Read out options */
313 while (options && options->name) {
314 if (!strcmp(options->name, BLOCK_OPT_SIZE)) {
315 bytes = options->value.n;
316 } else if (!strcmp(options->name, BLOCK_OPT_CLUSTER_SIZE)) {
317 if (options->value.n) {
318 objsize = options->value.n;
319 if ((objsize - 1) & objsize) { /* not a power of 2? */
320 error_report("obj size needs to be power of 2");
321 return -EINVAL;
323 if (objsize < 4096) {
324 error_report("obj size too small");
325 return -EINVAL;
327 obj_order = ffs(objsize) - 1;
330 options++;
333 clientname = qemu_rbd_parse_clientname(conf, clientname_buf);
334 if (rados_create(&cluster, clientname) < 0) {
335 error_report("error initializing");
336 return -EIO;
339 if (strstr(conf, "conf=") == NULL) {
340 /* try default location, but ignore failure */
341 rados_conf_read_file(cluster, NULL);
344 if (conf[0] != '\0' &&
345 qemu_rbd_set_conf(cluster, conf) < 0) {
346 error_report("error setting config options");
347 rados_shutdown(cluster);
348 return -EIO;
351 if (rados_connect(cluster) < 0) {
352 error_report("error connecting");
353 rados_shutdown(cluster);
354 return -EIO;
357 if (rados_ioctx_create(cluster, pool, &io_ctx) < 0) {
358 error_report("error opening pool %s", pool);
359 rados_shutdown(cluster);
360 return -EIO;
363 ret = rbd_create(io_ctx, name, bytes, &obj_order);
364 rados_ioctx_destroy(io_ctx);
365 rados_shutdown(cluster);
367 return ret;
371 * This aio completion is being called from qemu_rbd_aio_event_reader()
372 * and runs in qemu context. It schedules a bh, but just in case the aio
373 * was not cancelled before.
375 static void qemu_rbd_complete_aio(RADOSCB *rcb)
377 RBDAIOCB *acb = rcb->acb;
378 int64_t r;
380 r = rcb->ret;
382 if (acb->cmd == RBD_AIO_WRITE ||
383 acb->cmd == RBD_AIO_DISCARD) {
384 if (r < 0) {
385 acb->ret = r;
386 acb->error = 1;
387 } else if (!acb->error) {
388 acb->ret = rcb->size;
390 } else {
391 if (r < 0) {
392 memset(rcb->buf, 0, rcb->size);
393 acb->ret = r;
394 acb->error = 1;
395 } else if (r < rcb->size) {
396 memset(rcb->buf + r, 0, rcb->size - r);
397 if (!acb->error) {
398 acb->ret = rcb->size;
400 } else if (!acb->error) {
401 acb->ret = r;
404 /* Note that acb->bh can be NULL in case where the aio was cancelled */
405 acb->bh = qemu_bh_new(rbd_aio_bh_cb, acb);
406 qemu_bh_schedule(acb->bh);
407 g_free(rcb);
411 * aio fd read handler. It runs in the qemu context and calls the
412 * completion handling of completed rados aio operations.
414 static void qemu_rbd_aio_event_reader(void *opaque)
416 BDRVRBDState *s = opaque;
418 ssize_t ret;
420 do {
421 char *p = (char *)&s->event_rcb;
423 /* now read the rcb pointer that was sent from a non qemu thread */
424 ret = read(s->fds[RBD_FD_READ], p + s->event_reader_pos,
425 sizeof(s->event_rcb) - s->event_reader_pos);
426 if (ret > 0) {
427 s->event_reader_pos += ret;
428 if (s->event_reader_pos == sizeof(s->event_rcb)) {
429 s->event_reader_pos = 0;
430 qemu_rbd_complete_aio(s->event_rcb);
431 s->qemu_aio_count--;
434 } while (ret < 0 && errno == EINTR);
437 static int qemu_rbd_aio_flush_cb(void *opaque)
439 BDRVRBDState *s = opaque;
441 return (s->qemu_aio_count > 0);
444 static int qemu_rbd_open(BlockDriverState *bs, const char *filename, int flags)
446 BDRVRBDState *s = bs->opaque;
447 char pool[RBD_MAX_POOL_NAME_SIZE];
448 char snap_buf[RBD_MAX_SNAP_NAME_SIZE];
449 char conf[RBD_MAX_CONF_SIZE];
450 char clientname_buf[RBD_MAX_CONF_SIZE];
451 char *clientname;
452 int r;
454 if (qemu_rbd_parsename(filename, pool, sizeof(pool),
455 snap_buf, sizeof(snap_buf),
456 s->name, sizeof(s->name),
457 conf, sizeof(conf)) < 0) {
458 return -EINVAL;
461 clientname = qemu_rbd_parse_clientname(conf, clientname_buf);
462 r = rados_create(&s->cluster, clientname);
463 if (r < 0) {
464 error_report("error initializing");
465 return r;
468 s->snap = NULL;
469 if (snap_buf[0] != '\0') {
470 s->snap = g_strdup(snap_buf);
474 * Fallback to more conservative semantics if setting cache
475 * options fails. Ignore errors from setting rbd_cache because the
476 * only possible error is that the option does not exist, and
477 * librbd defaults to no caching. If write through caching cannot
478 * be set up, fall back to no caching.
480 if (flags & BDRV_O_NOCACHE) {
481 rados_conf_set(s->cluster, "rbd_cache", "false");
482 } else {
483 rados_conf_set(s->cluster, "rbd_cache", "true");
486 if (strstr(conf, "conf=") == NULL) {
487 /* try default location, but ignore failure */
488 rados_conf_read_file(s->cluster, NULL);
491 if (conf[0] != '\0') {
492 r = qemu_rbd_set_conf(s->cluster, conf);
493 if (r < 0) {
494 error_report("error setting config options");
495 goto failed_shutdown;
499 r = rados_connect(s->cluster);
500 if (r < 0) {
501 error_report("error connecting");
502 goto failed_shutdown;
505 r = rados_ioctx_create(s->cluster, pool, &s->io_ctx);
506 if (r < 0) {
507 error_report("error opening pool %s", pool);
508 goto failed_shutdown;
511 r = rbd_open(s->io_ctx, s->name, &s->image, s->snap);
512 if (r < 0) {
513 error_report("error reading header from %s", s->name);
514 goto failed_open;
517 bs->read_only = (s->snap != NULL);
519 s->event_reader_pos = 0;
520 r = qemu_pipe(s->fds);
521 if (r < 0) {
522 error_report("error opening eventfd");
523 goto failed;
525 fcntl(s->fds[0], F_SETFL, O_NONBLOCK);
526 fcntl(s->fds[1], F_SETFL, O_NONBLOCK);
527 qemu_aio_set_fd_handler(s->fds[RBD_FD_READ], qemu_rbd_aio_event_reader,
528 NULL, qemu_rbd_aio_flush_cb, s);
531 return 0;
533 failed:
534 rbd_close(s->image);
535 failed_open:
536 rados_ioctx_destroy(s->io_ctx);
537 failed_shutdown:
538 rados_shutdown(s->cluster);
539 g_free(s->snap);
540 return r;
543 static void qemu_rbd_close(BlockDriverState *bs)
545 BDRVRBDState *s = bs->opaque;
547 close(s->fds[0]);
548 close(s->fds[1]);
549 qemu_aio_set_fd_handler(s->fds[RBD_FD_READ], NULL, NULL, NULL, NULL);
551 rbd_close(s->image);
552 rados_ioctx_destroy(s->io_ctx);
553 g_free(s->snap);
554 rados_shutdown(s->cluster);
558 * Cancel aio. Since we don't reference acb in a non qemu threads,
559 * it is safe to access it here.
561 static void qemu_rbd_aio_cancel(BlockDriverAIOCB *blockacb)
563 RBDAIOCB *acb = (RBDAIOCB *) blockacb;
564 acb->cancelled = 1;
566 while (acb->status == -EINPROGRESS) {
567 qemu_aio_wait();
570 qemu_aio_release(acb);
573 static const AIOCBInfo rbd_aiocb_info = {
574 .aiocb_size = sizeof(RBDAIOCB),
575 .cancel = qemu_rbd_aio_cancel,
578 static int qemu_rbd_send_pipe(BDRVRBDState *s, RADOSCB *rcb)
580 int ret = 0;
581 while (1) {
582 fd_set wfd;
583 int fd = s->fds[RBD_FD_WRITE];
585 /* send the op pointer to the qemu thread that is responsible
586 for the aio/op completion. Must do it in a qemu thread context */
587 ret = write(fd, (void *)&rcb, sizeof(rcb));
588 if (ret >= 0) {
589 break;
591 if (errno == EINTR) {
592 continue;
594 if (errno != EAGAIN) {
595 break;
598 FD_ZERO(&wfd);
599 FD_SET(fd, &wfd);
600 do {
601 ret = select(fd + 1, NULL, &wfd, NULL, NULL);
602 } while (ret < 0 && errno == EINTR);
605 return ret;
609 * This is the callback function for rbd_aio_read and _write
611 * Note: this function is being called from a non qemu thread so
612 * we need to be careful about what we do here. Generally we only
613 * write to the block notification pipe, and do the rest of the
614 * io completion handling from qemu_rbd_aio_event_reader() which
615 * runs in a qemu context.
617 static void rbd_finish_aiocb(rbd_completion_t c, RADOSCB *rcb)
619 int ret;
620 rcb->ret = rbd_aio_get_return_value(c);
621 rbd_aio_release(c);
622 ret = qemu_rbd_send_pipe(rcb->s, rcb);
623 if (ret < 0) {
624 error_report("failed writing to acb->s->fds");
625 g_free(rcb);
629 /* Callback when all queued rbd_aio requests are complete */
631 static void rbd_aio_bh_cb(void *opaque)
633 RBDAIOCB *acb = opaque;
635 if (acb->cmd == RBD_AIO_READ) {
636 qemu_iovec_from_buf(acb->qiov, 0, acb->bounce, acb->qiov->size);
638 qemu_vfree(acb->bounce);
639 acb->common.cb(acb->common.opaque, (acb->ret > 0 ? 0 : acb->ret));
640 qemu_bh_delete(acb->bh);
641 acb->bh = NULL;
642 acb->status = 0;
644 if (!acb->cancelled) {
645 qemu_aio_release(acb);
649 static int rbd_aio_discard_wrapper(rbd_image_t image,
650 uint64_t off,
651 uint64_t len,
652 rbd_completion_t comp)
654 #ifdef LIBRBD_SUPPORTS_DISCARD
655 return rbd_aio_discard(image, off, len, comp);
656 #else
657 return -ENOTSUP;
658 #endif
661 static BlockDriverAIOCB *rbd_start_aio(BlockDriverState *bs,
662 int64_t sector_num,
663 QEMUIOVector *qiov,
664 int nb_sectors,
665 BlockDriverCompletionFunc *cb,
666 void *opaque,
667 RBDAIOCmd cmd)
669 RBDAIOCB *acb;
670 RADOSCB *rcb;
671 rbd_completion_t c;
672 int64_t off, size;
673 char *buf;
674 int r;
676 BDRVRBDState *s = bs->opaque;
678 acb = qemu_aio_get(&rbd_aiocb_info, bs, cb, opaque);
679 acb->cmd = cmd;
680 acb->qiov = qiov;
681 if (cmd == RBD_AIO_DISCARD) {
682 acb->bounce = NULL;
683 } else {
684 acb->bounce = qemu_blockalign(bs, qiov->size);
686 acb->ret = 0;
687 acb->error = 0;
688 acb->s = s;
689 acb->cancelled = 0;
690 acb->bh = NULL;
691 acb->status = -EINPROGRESS;
693 if (cmd == RBD_AIO_WRITE) {
694 qemu_iovec_to_buf(acb->qiov, 0, acb->bounce, qiov->size);
697 buf = acb->bounce;
699 off = sector_num * BDRV_SECTOR_SIZE;
700 size = nb_sectors * BDRV_SECTOR_SIZE;
702 s->qemu_aio_count++; /* All the RADOSCB */
704 rcb = g_malloc(sizeof(RADOSCB));
705 rcb->done = 0;
706 rcb->acb = acb;
707 rcb->buf = buf;
708 rcb->s = acb->s;
709 rcb->size = size;
710 r = rbd_aio_create_completion(rcb, (rbd_callback_t) rbd_finish_aiocb, &c);
711 if (r < 0) {
712 goto failed;
715 switch (cmd) {
716 case RBD_AIO_WRITE:
717 r = rbd_aio_write(s->image, off, size, buf, c);
718 break;
719 case RBD_AIO_READ:
720 r = rbd_aio_read(s->image, off, size, buf, c);
721 break;
722 case RBD_AIO_DISCARD:
723 r = rbd_aio_discard_wrapper(s->image, off, size, c);
724 break;
725 default:
726 r = -EINVAL;
729 if (r < 0) {
730 goto failed;
733 return &acb->common;
735 failed:
736 g_free(rcb);
737 s->qemu_aio_count--;
738 qemu_aio_release(acb);
739 return NULL;
742 static BlockDriverAIOCB *qemu_rbd_aio_readv(BlockDriverState *bs,
743 int64_t sector_num,
744 QEMUIOVector *qiov,
745 int nb_sectors,
746 BlockDriverCompletionFunc *cb,
747 void *opaque)
749 return rbd_start_aio(bs, sector_num, qiov, nb_sectors, cb, opaque,
750 RBD_AIO_READ);
753 static BlockDriverAIOCB *qemu_rbd_aio_writev(BlockDriverState *bs,
754 int64_t sector_num,
755 QEMUIOVector *qiov,
756 int nb_sectors,
757 BlockDriverCompletionFunc *cb,
758 void *opaque)
760 return rbd_start_aio(bs, sector_num, qiov, nb_sectors, cb, opaque,
761 RBD_AIO_WRITE);
764 static int qemu_rbd_co_flush(BlockDriverState *bs)
766 #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 1)
767 /* rbd_flush added in 0.1.1 */
768 BDRVRBDState *s = bs->opaque;
769 return rbd_flush(s->image);
770 #else
771 return 0;
772 #endif
775 static int qemu_rbd_getinfo(BlockDriverState *bs, BlockDriverInfo *bdi)
777 BDRVRBDState *s = bs->opaque;
778 rbd_image_info_t info;
779 int r;
781 r = rbd_stat(s->image, &info, sizeof(info));
782 if (r < 0) {
783 return r;
786 bdi->cluster_size = info.obj_size;
787 return 0;
790 static int64_t qemu_rbd_getlength(BlockDriverState *bs)
792 BDRVRBDState *s = bs->opaque;
793 rbd_image_info_t info;
794 int r;
796 r = rbd_stat(s->image, &info, sizeof(info));
797 if (r < 0) {
798 return r;
801 return info.size;
804 static int qemu_rbd_truncate(BlockDriverState *bs, int64_t offset)
806 BDRVRBDState *s = bs->opaque;
807 int r;
809 r = rbd_resize(s->image, offset);
810 if (r < 0) {
811 return r;
814 return 0;
817 static int qemu_rbd_snap_create(BlockDriverState *bs,
818 QEMUSnapshotInfo *sn_info)
820 BDRVRBDState *s = bs->opaque;
821 int r;
823 if (sn_info->name[0] == '\0') {
824 return -EINVAL; /* we need a name for rbd snapshots */
828 * rbd snapshots are using the name as the user controlled unique identifier
829 * we can't use the rbd snapid for that purpose, as it can't be set
831 if (sn_info->id_str[0] != '\0' &&
832 strcmp(sn_info->id_str, sn_info->name) != 0) {
833 return -EINVAL;
836 if (strlen(sn_info->name) >= sizeof(sn_info->id_str)) {
837 return -ERANGE;
840 r = rbd_snap_create(s->image, sn_info->name);
841 if (r < 0) {
842 error_report("failed to create snap: %s", strerror(-r));
843 return r;
846 return 0;
849 static int qemu_rbd_snap_remove(BlockDriverState *bs,
850 const char *snapshot_name)
852 BDRVRBDState *s = bs->opaque;
853 int r;
855 r = rbd_snap_remove(s->image, snapshot_name);
856 return r;
859 static int qemu_rbd_snap_rollback(BlockDriverState *bs,
860 const char *snapshot_name)
862 BDRVRBDState *s = bs->opaque;
863 int r;
865 r = rbd_snap_rollback(s->image, snapshot_name);
866 return r;
869 static int qemu_rbd_snap_list(BlockDriverState *bs,
870 QEMUSnapshotInfo **psn_tab)
872 BDRVRBDState *s = bs->opaque;
873 QEMUSnapshotInfo *sn_info, *sn_tab = NULL;
874 int i, snap_count;
875 rbd_snap_info_t *snaps;
876 int max_snaps = RBD_MAX_SNAPS;
878 do {
879 snaps = g_malloc(sizeof(*snaps) * max_snaps);
880 snap_count = rbd_snap_list(s->image, snaps, &max_snaps);
881 if (snap_count < 0) {
882 g_free(snaps);
884 } while (snap_count == -ERANGE);
886 if (snap_count <= 0) {
887 goto done;
890 sn_tab = g_malloc0(snap_count * sizeof(QEMUSnapshotInfo));
892 for (i = 0; i < snap_count; i++) {
893 const char *snap_name = snaps[i].name;
895 sn_info = sn_tab + i;
896 pstrcpy(sn_info->id_str, sizeof(sn_info->id_str), snap_name);
897 pstrcpy(sn_info->name, sizeof(sn_info->name), snap_name);
899 sn_info->vm_state_size = snaps[i].size;
900 sn_info->date_sec = 0;
901 sn_info->date_nsec = 0;
902 sn_info->vm_clock_nsec = 0;
904 rbd_snap_list_end(snaps);
906 done:
907 *psn_tab = sn_tab;
908 return snap_count;
911 #ifdef LIBRBD_SUPPORTS_DISCARD
912 static BlockDriverAIOCB* qemu_rbd_aio_discard(BlockDriverState *bs,
913 int64_t sector_num,
914 int nb_sectors,
915 BlockDriverCompletionFunc *cb,
916 void *opaque)
918 return rbd_start_aio(bs, sector_num, NULL, nb_sectors, cb, opaque,
919 RBD_AIO_DISCARD);
921 #endif
923 static QEMUOptionParameter qemu_rbd_create_options[] = {
925 .name = BLOCK_OPT_SIZE,
926 .type = OPT_SIZE,
927 .help = "Virtual disk size"
930 .name = BLOCK_OPT_CLUSTER_SIZE,
931 .type = OPT_SIZE,
932 .help = "RBD object size"
934 {NULL}
937 static BlockDriver bdrv_rbd = {
938 .format_name = "rbd",
939 .instance_size = sizeof(BDRVRBDState),
940 .bdrv_file_open = qemu_rbd_open,
941 .bdrv_close = qemu_rbd_close,
942 .bdrv_create = qemu_rbd_create,
943 .bdrv_get_info = qemu_rbd_getinfo,
944 .create_options = qemu_rbd_create_options,
945 .bdrv_getlength = qemu_rbd_getlength,
946 .bdrv_truncate = qemu_rbd_truncate,
947 .protocol_name = "rbd",
949 .bdrv_aio_readv = qemu_rbd_aio_readv,
950 .bdrv_aio_writev = qemu_rbd_aio_writev,
951 .bdrv_co_flush_to_disk = qemu_rbd_co_flush,
953 #ifdef LIBRBD_SUPPORTS_DISCARD
954 .bdrv_aio_discard = qemu_rbd_aio_discard,
955 #endif
957 .bdrv_snapshot_create = qemu_rbd_snap_create,
958 .bdrv_snapshot_delete = qemu_rbd_snap_remove,
959 .bdrv_snapshot_list = qemu_rbd_snap_list,
960 .bdrv_snapshot_goto = qemu_rbd_snap_rollback,
963 static void bdrv_rbd_init(void)
965 bdrv_register(&bdrv_rbd);
968 block_init(bdrv_rbd_init);