blockdev: Error out on negative throttling option values
[qemu/ar7.git] / block / rbd.c
blob51b64f3fed7e3e3fd992ad3fd4fefe53a48373f7
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 "qemu/osdep.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 RBD_AIO_FLUSH
68 } RBDAIOCmd;
70 typedef struct RBDAIOCB {
71 BlockAIOCB common;
72 QEMUBH *bh;
73 int64_t ret;
74 QEMUIOVector *qiov;
75 char *bounce;
76 RBDAIOCmd cmd;
77 int error;
78 struct BDRVRBDState *s;
79 } RBDAIOCB;
81 typedef struct RADOSCB {
82 RBDAIOCB *acb;
83 struct BDRVRBDState *s;
84 int64_t size;
85 char *buf;
86 int64_t ret;
87 } RADOSCB;
89 typedef struct BDRVRBDState {
90 rados_t cluster;
91 rados_ioctx_t io_ctx;
92 rbd_image_t image;
93 char name[RBD_MAX_IMAGE_NAME_SIZE];
94 char *snap;
95 } BDRVRBDState;
97 static int qemu_rbd_next_tok(char *dst, int dst_len,
98 char *src, char delim,
99 const char *name,
100 char **p, Error **errp)
102 int l;
103 char *end;
105 *p = NULL;
107 if (delim != '\0') {
108 for (end = src; *end; ++end) {
109 if (*end == delim) {
110 break;
112 if (*end == '\\' && end[1] != '\0') {
113 end++;
116 if (*end == delim) {
117 *p = end + 1;
118 *end = '\0';
121 l = strlen(src);
122 if (l >= dst_len) {
123 error_setg(errp, "%s too long", name);
124 return -EINVAL;
125 } else if (l == 0) {
126 error_setg(errp, "%s too short", name);
127 return -EINVAL;
130 pstrcpy(dst, dst_len, src);
132 return 0;
135 static void qemu_rbd_unescape(char *src)
137 char *p;
139 for (p = src; *src; ++src, ++p) {
140 if (*src == '\\' && src[1] != '\0') {
141 src++;
143 *p = *src;
145 *p = '\0';
148 static int qemu_rbd_parsename(const char *filename,
149 char *pool, int pool_len,
150 char *snap, int snap_len,
151 char *name, int name_len,
152 char *conf, int conf_len,
153 Error **errp)
155 const char *start;
156 char *p, *buf;
157 int ret;
159 if (!strstart(filename, "rbd:", &start)) {
160 error_setg(errp, "File name must start with 'rbd:'");
161 return -EINVAL;
164 buf = g_strdup(start);
165 p = buf;
166 *snap = '\0';
167 *conf = '\0';
169 ret = qemu_rbd_next_tok(pool, pool_len, p,
170 '/', "pool name", &p, errp);
171 if (ret < 0 || !p) {
172 ret = -EINVAL;
173 goto done;
175 qemu_rbd_unescape(pool);
177 if (strchr(p, '@')) {
178 ret = qemu_rbd_next_tok(name, name_len, p,
179 '@', "object name", &p, errp);
180 if (ret < 0) {
181 goto done;
183 ret = qemu_rbd_next_tok(snap, snap_len, p,
184 ':', "snap name", &p, errp);
185 qemu_rbd_unescape(snap);
186 } else {
187 ret = qemu_rbd_next_tok(name, name_len, p,
188 ':', "object name", &p, errp);
190 qemu_rbd_unescape(name);
191 if (ret < 0 || !p) {
192 goto done;
195 ret = qemu_rbd_next_tok(conf, conf_len, p,
196 '\0', "configuration", &p, errp);
198 done:
199 g_free(buf);
200 return ret;
203 static char *qemu_rbd_parse_clientname(const char *conf, char *clientname)
205 const char *p = conf;
207 while (*p) {
208 int len;
209 const char *end = strchr(p, ':');
211 if (end) {
212 len = end - p;
213 } else {
214 len = strlen(p);
217 if (strncmp(p, "id=", 3) == 0) {
218 len -= 3;
219 strncpy(clientname, p + 3, len);
220 clientname[len] = '\0';
221 return clientname;
223 if (end == NULL) {
224 break;
226 p = end + 1;
228 return NULL;
231 static int qemu_rbd_set_conf(rados_t cluster, const char *conf,
232 bool only_read_conf_file,
233 Error **errp)
235 char *p, *buf;
236 char name[RBD_MAX_CONF_NAME_SIZE];
237 char value[RBD_MAX_CONF_VAL_SIZE];
238 int ret = 0;
240 buf = g_strdup(conf);
241 p = buf;
243 while (p) {
244 ret = qemu_rbd_next_tok(name, sizeof(name), p,
245 '=', "conf option name", &p, errp);
246 if (ret < 0) {
247 break;
249 qemu_rbd_unescape(name);
251 if (!p) {
252 error_setg(errp, "conf option %s has no value", name);
253 ret = -EINVAL;
254 break;
257 ret = qemu_rbd_next_tok(value, sizeof(value), p,
258 ':', "conf option value", &p, errp);
259 if (ret < 0) {
260 break;
262 qemu_rbd_unescape(value);
264 if (strcmp(name, "conf") == 0) {
265 /* read the conf file alone, so it doesn't override more
266 specific settings for a particular device */
267 if (only_read_conf_file) {
268 ret = rados_conf_read_file(cluster, value);
269 if (ret < 0) {
270 error_setg(errp, "error reading conf file %s", value);
271 break;
274 } else if (strcmp(name, "id") == 0) {
275 /* ignore, this is parsed by qemu_rbd_parse_clientname() */
276 } else if (!only_read_conf_file) {
277 ret = rados_conf_set(cluster, name, value);
278 if (ret < 0) {
279 error_setg(errp, "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, QemuOpts *opts, Error **errp)
292 Error *local_err = NULL;
293 int64_t bytes = 0;
294 int64_t objsize;
295 int obj_order = 0;
296 char pool[RBD_MAX_POOL_NAME_SIZE];
297 char name[RBD_MAX_IMAGE_NAME_SIZE];
298 char snap_buf[RBD_MAX_SNAP_NAME_SIZE];
299 char conf[RBD_MAX_CONF_SIZE];
300 char clientname_buf[RBD_MAX_CONF_SIZE];
301 char *clientname;
302 rados_t cluster;
303 rados_ioctx_t io_ctx;
304 int ret;
306 if (qemu_rbd_parsename(filename, pool, sizeof(pool),
307 snap_buf, sizeof(snap_buf),
308 name, sizeof(name),
309 conf, sizeof(conf), &local_err) < 0) {
310 error_propagate(errp, local_err);
311 return -EINVAL;
314 /* Read out options */
315 bytes = ROUND_UP(qemu_opt_get_size_del(opts, BLOCK_OPT_SIZE, 0),
316 BDRV_SECTOR_SIZE);
317 objsize = qemu_opt_get_size_del(opts, BLOCK_OPT_CLUSTER_SIZE, 0);
318 if (objsize) {
319 if ((objsize - 1) & objsize) { /* not a power of 2? */
320 error_setg(errp, "obj size needs to be power of 2");
321 return -EINVAL;
323 if (objsize < 4096) {
324 error_setg(errp, "obj size too small");
325 return -EINVAL;
327 obj_order = ctz32(objsize);
330 clientname = qemu_rbd_parse_clientname(conf, clientname_buf);
331 if (rados_create(&cluster, clientname) < 0) {
332 error_setg(errp, "error initializing");
333 return -EIO;
336 if (strstr(conf, "conf=") == NULL) {
337 /* try default location, but ignore failure */
338 rados_conf_read_file(cluster, NULL);
339 } else if (conf[0] != '\0' &&
340 qemu_rbd_set_conf(cluster, conf, true, &local_err) < 0) {
341 rados_shutdown(cluster);
342 error_propagate(errp, local_err);
343 return -EIO;
346 if (conf[0] != '\0' &&
347 qemu_rbd_set_conf(cluster, conf, false, &local_err) < 0) {
348 rados_shutdown(cluster);
349 error_propagate(errp, local_err);
350 return -EIO;
353 if (rados_connect(cluster) < 0) {
354 error_setg(errp, "error connecting");
355 rados_shutdown(cluster);
356 return -EIO;
359 if (rados_ioctx_create(cluster, pool, &io_ctx) < 0) {
360 error_setg(errp, "error opening pool %s", pool);
361 rados_shutdown(cluster);
362 return -EIO;
365 ret = rbd_create(io_ctx, name, bytes, &obj_order);
366 rados_ioctx_destroy(io_ctx);
367 rados_shutdown(cluster);
369 return ret;
373 * This aio completion is being called from rbd_finish_bh() and runs in qemu
374 * BH context.
376 static void qemu_rbd_complete_aio(RADOSCB *rcb)
378 RBDAIOCB *acb = rcb->acb;
379 int64_t r;
381 r = rcb->ret;
383 if (acb->cmd != RBD_AIO_READ) {
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;
405 g_free(rcb);
407 if (acb->cmd == RBD_AIO_READ) {
408 qemu_iovec_from_buf(acb->qiov, 0, acb->bounce, acb->qiov->size);
410 qemu_vfree(acb->bounce);
411 acb->common.cb(acb->common.opaque, (acb->ret > 0 ? 0 : acb->ret));
413 qemu_aio_unref(acb);
416 /* TODO Convert to fine grained options */
417 static QemuOptsList runtime_opts = {
418 .name = "rbd",
419 .head = QTAILQ_HEAD_INITIALIZER(runtime_opts.head),
420 .desc = {
422 .name = "filename",
423 .type = QEMU_OPT_STRING,
424 .help = "Specification of the rbd image",
426 { /* end of list */ }
430 static int qemu_rbd_open(BlockDriverState *bs, QDict *options, int flags,
431 Error **errp)
433 BDRVRBDState *s = bs->opaque;
434 char pool[RBD_MAX_POOL_NAME_SIZE];
435 char snap_buf[RBD_MAX_SNAP_NAME_SIZE];
436 char conf[RBD_MAX_CONF_SIZE];
437 char clientname_buf[RBD_MAX_CONF_SIZE];
438 char *clientname;
439 QemuOpts *opts;
440 Error *local_err = NULL;
441 const char *filename;
442 int r;
444 opts = qemu_opts_create(&runtime_opts, NULL, 0, &error_abort);
445 qemu_opts_absorb_qdict(opts, options, &local_err);
446 if (local_err) {
447 error_propagate(errp, local_err);
448 qemu_opts_del(opts);
449 return -EINVAL;
452 filename = qemu_opt_get(opts, "filename");
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), errp) < 0) {
458 r = -EINVAL;
459 goto failed_opts;
462 clientname = qemu_rbd_parse_clientname(conf, clientname_buf);
463 r = rados_create(&s->cluster, clientname);
464 if (r < 0) {
465 error_setg(errp, "error initializing");
466 goto failed_opts;
469 s->snap = NULL;
470 if (snap_buf[0] != '\0') {
471 s->snap = g_strdup(snap_buf);
474 if (strstr(conf, "conf=") == NULL) {
475 /* try default location, but ignore failure */
476 rados_conf_read_file(s->cluster, NULL);
477 } else if (conf[0] != '\0') {
478 r = qemu_rbd_set_conf(s->cluster, conf, true, errp);
479 if (r < 0) {
480 goto failed_shutdown;
484 if (conf[0] != '\0') {
485 r = qemu_rbd_set_conf(s->cluster, conf, false, errp);
486 if (r < 0) {
487 goto failed_shutdown;
492 * Fallback to more conservative semantics if setting cache
493 * options fails. Ignore errors from setting rbd_cache because the
494 * only possible error is that the option does not exist, and
495 * librbd defaults to no caching. If write through caching cannot
496 * be set up, fall back to no caching.
498 if (flags & BDRV_O_NOCACHE) {
499 rados_conf_set(s->cluster, "rbd_cache", "false");
500 } else {
501 rados_conf_set(s->cluster, "rbd_cache", "true");
504 r = rados_connect(s->cluster);
505 if (r < 0) {
506 error_setg(errp, "error connecting");
507 goto failed_shutdown;
510 r = rados_ioctx_create(s->cluster, pool, &s->io_ctx);
511 if (r < 0) {
512 error_setg(errp, "error opening pool %s", pool);
513 goto failed_shutdown;
516 r = rbd_open(s->io_ctx, s->name, &s->image, s->snap);
517 if (r < 0) {
518 error_setg(errp, "error reading header from %s", s->name);
519 goto failed_open;
522 bs->read_only = (s->snap != NULL);
524 qemu_opts_del(opts);
525 return 0;
527 failed_open:
528 rados_ioctx_destroy(s->io_ctx);
529 failed_shutdown:
530 rados_shutdown(s->cluster);
531 g_free(s->snap);
532 failed_opts:
533 qemu_opts_del(opts);
534 return r;
537 static void qemu_rbd_close(BlockDriverState *bs)
539 BDRVRBDState *s = bs->opaque;
541 rbd_close(s->image);
542 rados_ioctx_destroy(s->io_ctx);
543 g_free(s->snap);
544 rados_shutdown(s->cluster);
547 static const AIOCBInfo rbd_aiocb_info = {
548 .aiocb_size = sizeof(RBDAIOCB),
551 static void rbd_finish_bh(void *opaque)
553 RADOSCB *rcb = opaque;
554 qemu_bh_delete(rcb->acb->bh);
555 qemu_rbd_complete_aio(rcb);
559 * This is the callback function for rbd_aio_read and _write
561 * Note: this function is being called from a non qemu thread so
562 * we need to be careful about what we do here. Generally we only
563 * schedule a BH, and do the rest of the io completion handling
564 * from rbd_finish_bh() which runs in a qemu context.
566 static void rbd_finish_aiocb(rbd_completion_t c, RADOSCB *rcb)
568 RBDAIOCB *acb = rcb->acb;
570 rcb->ret = rbd_aio_get_return_value(c);
571 rbd_aio_release(c);
573 acb->bh = aio_bh_new(bdrv_get_aio_context(acb->common.bs),
574 rbd_finish_bh, rcb);
575 qemu_bh_schedule(acb->bh);
578 static int rbd_aio_discard_wrapper(rbd_image_t image,
579 uint64_t off,
580 uint64_t len,
581 rbd_completion_t comp)
583 #ifdef LIBRBD_SUPPORTS_DISCARD
584 return rbd_aio_discard(image, off, len, comp);
585 #else
586 return -ENOTSUP;
587 #endif
590 static int rbd_aio_flush_wrapper(rbd_image_t image,
591 rbd_completion_t comp)
593 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
594 return rbd_aio_flush(image, comp);
595 #else
596 return -ENOTSUP;
597 #endif
600 static BlockAIOCB *rbd_start_aio(BlockDriverState *bs,
601 int64_t sector_num,
602 QEMUIOVector *qiov,
603 int nb_sectors,
604 BlockCompletionFunc *cb,
605 void *opaque,
606 RBDAIOCmd cmd)
608 RBDAIOCB *acb;
609 RADOSCB *rcb = NULL;
610 rbd_completion_t c;
611 int64_t off, size;
612 char *buf;
613 int r;
615 BDRVRBDState *s = bs->opaque;
617 acb = qemu_aio_get(&rbd_aiocb_info, bs, cb, opaque);
618 acb->cmd = cmd;
619 acb->qiov = qiov;
620 if (cmd == RBD_AIO_DISCARD || cmd == RBD_AIO_FLUSH) {
621 acb->bounce = NULL;
622 } else {
623 acb->bounce = qemu_try_blockalign(bs, qiov->size);
624 if (acb->bounce == NULL) {
625 goto failed;
628 acb->ret = 0;
629 acb->error = 0;
630 acb->s = s;
631 acb->bh = NULL;
633 if (cmd == RBD_AIO_WRITE) {
634 qemu_iovec_to_buf(acb->qiov, 0, acb->bounce, qiov->size);
637 buf = acb->bounce;
639 off = sector_num * BDRV_SECTOR_SIZE;
640 size = nb_sectors * BDRV_SECTOR_SIZE;
642 rcb = g_new(RADOSCB, 1);
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;
652 switch (cmd) {
653 case RBD_AIO_WRITE:
654 r = rbd_aio_write(s->image, off, size, buf, c);
655 break;
656 case RBD_AIO_READ:
657 r = rbd_aio_read(s->image, off, size, buf, c);
658 break;
659 case RBD_AIO_DISCARD:
660 r = rbd_aio_discard_wrapper(s->image, off, size, c);
661 break;
662 case RBD_AIO_FLUSH:
663 r = rbd_aio_flush_wrapper(s->image, c);
664 break;
665 default:
666 r = -EINVAL;
669 if (r < 0) {
670 goto failed_completion;
673 return &acb->common;
675 failed_completion:
676 rbd_aio_release(c);
677 failed:
678 g_free(rcb);
679 qemu_vfree(acb->bounce);
680 qemu_aio_unref(acb);
681 return NULL;
684 static BlockAIOCB *qemu_rbd_aio_readv(BlockDriverState *bs,
685 int64_t sector_num,
686 QEMUIOVector *qiov,
687 int nb_sectors,
688 BlockCompletionFunc *cb,
689 void *opaque)
691 return rbd_start_aio(bs, sector_num, qiov, nb_sectors, cb, opaque,
692 RBD_AIO_READ);
695 static BlockAIOCB *qemu_rbd_aio_writev(BlockDriverState *bs,
696 int64_t sector_num,
697 QEMUIOVector *qiov,
698 int nb_sectors,
699 BlockCompletionFunc *cb,
700 void *opaque)
702 return rbd_start_aio(bs, sector_num, qiov, nb_sectors, cb, opaque,
703 RBD_AIO_WRITE);
706 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
707 static BlockAIOCB *qemu_rbd_aio_flush(BlockDriverState *bs,
708 BlockCompletionFunc *cb,
709 void *opaque)
711 return rbd_start_aio(bs, 0, NULL, 0, cb, opaque, RBD_AIO_FLUSH);
714 #else
716 static int qemu_rbd_co_flush(BlockDriverState *bs)
718 #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 1)
719 /* rbd_flush added in 0.1.1 */
720 BDRVRBDState *s = bs->opaque;
721 return rbd_flush(s->image);
722 #else
723 return 0;
724 #endif
726 #endif
728 static int qemu_rbd_getinfo(BlockDriverState *bs, BlockDriverInfo *bdi)
730 BDRVRBDState *s = bs->opaque;
731 rbd_image_info_t info;
732 int r;
734 r = rbd_stat(s->image, &info, sizeof(info));
735 if (r < 0) {
736 return r;
739 bdi->cluster_size = info.obj_size;
740 return 0;
743 static int64_t qemu_rbd_getlength(BlockDriverState *bs)
745 BDRVRBDState *s = bs->opaque;
746 rbd_image_info_t info;
747 int r;
749 r = rbd_stat(s->image, &info, sizeof(info));
750 if (r < 0) {
751 return r;
754 return info.size;
757 static int qemu_rbd_truncate(BlockDriverState *bs, int64_t offset)
759 BDRVRBDState *s = bs->opaque;
760 int r;
762 r = rbd_resize(s->image, offset);
763 if (r < 0) {
764 return r;
767 return 0;
770 static int qemu_rbd_snap_create(BlockDriverState *bs,
771 QEMUSnapshotInfo *sn_info)
773 BDRVRBDState *s = bs->opaque;
774 int r;
776 if (sn_info->name[0] == '\0') {
777 return -EINVAL; /* we need a name for rbd snapshots */
781 * rbd snapshots are using the name as the user controlled unique identifier
782 * we can't use the rbd snapid for that purpose, as it can't be set
784 if (sn_info->id_str[0] != '\0' &&
785 strcmp(sn_info->id_str, sn_info->name) != 0) {
786 return -EINVAL;
789 if (strlen(sn_info->name) >= sizeof(sn_info->id_str)) {
790 return -ERANGE;
793 r = rbd_snap_create(s->image, sn_info->name);
794 if (r < 0) {
795 error_report("failed to create snap: %s", strerror(-r));
796 return r;
799 return 0;
802 static int qemu_rbd_snap_remove(BlockDriverState *bs,
803 const char *snapshot_id,
804 const char *snapshot_name,
805 Error **errp)
807 BDRVRBDState *s = bs->opaque;
808 int r;
810 if (!snapshot_name) {
811 error_setg(errp, "rbd need a valid snapshot name");
812 return -EINVAL;
815 /* If snapshot_id is specified, it must be equal to name, see
816 qemu_rbd_snap_list() */
817 if (snapshot_id && strcmp(snapshot_id, snapshot_name)) {
818 error_setg(errp,
819 "rbd do not support snapshot id, it should be NULL or "
820 "equal to snapshot name");
821 return -EINVAL;
824 r = rbd_snap_remove(s->image, snapshot_name);
825 if (r < 0) {
826 error_setg_errno(errp, -r, "Failed to remove the snapshot");
828 return r;
831 static int qemu_rbd_snap_rollback(BlockDriverState *bs,
832 const char *snapshot_name)
834 BDRVRBDState *s = bs->opaque;
835 int r;
837 r = rbd_snap_rollback(s->image, snapshot_name);
838 return r;
841 static int qemu_rbd_snap_list(BlockDriverState *bs,
842 QEMUSnapshotInfo **psn_tab)
844 BDRVRBDState *s = bs->opaque;
845 QEMUSnapshotInfo *sn_info, *sn_tab = NULL;
846 int i, snap_count;
847 rbd_snap_info_t *snaps;
848 int max_snaps = RBD_MAX_SNAPS;
850 do {
851 snaps = g_new(rbd_snap_info_t, max_snaps);
852 snap_count = rbd_snap_list(s->image, snaps, &max_snaps);
853 if (snap_count <= 0) {
854 g_free(snaps);
856 } while (snap_count == -ERANGE);
858 if (snap_count <= 0) {
859 goto done;
862 sn_tab = g_new0(QEMUSnapshotInfo, snap_count);
864 for (i = 0; i < snap_count; i++) {
865 const char *snap_name = snaps[i].name;
867 sn_info = sn_tab + i;
868 pstrcpy(sn_info->id_str, sizeof(sn_info->id_str), snap_name);
869 pstrcpy(sn_info->name, sizeof(sn_info->name), snap_name);
871 sn_info->vm_state_size = snaps[i].size;
872 sn_info->date_sec = 0;
873 sn_info->date_nsec = 0;
874 sn_info->vm_clock_nsec = 0;
876 rbd_snap_list_end(snaps);
877 g_free(snaps);
879 done:
880 *psn_tab = sn_tab;
881 return snap_count;
884 #ifdef LIBRBD_SUPPORTS_DISCARD
885 static BlockAIOCB* qemu_rbd_aio_discard(BlockDriverState *bs,
886 int64_t sector_num,
887 int nb_sectors,
888 BlockCompletionFunc *cb,
889 void *opaque)
891 return rbd_start_aio(bs, sector_num, NULL, nb_sectors, cb, opaque,
892 RBD_AIO_DISCARD);
894 #endif
896 #ifdef LIBRBD_SUPPORTS_INVALIDATE
897 static void qemu_rbd_invalidate_cache(BlockDriverState *bs,
898 Error **errp)
900 BDRVRBDState *s = bs->opaque;
901 int r = rbd_invalidate_cache(s->image);
902 if (r < 0) {
903 error_setg_errno(errp, -r, "Failed to invalidate the cache");
906 #endif
908 static QemuOptsList qemu_rbd_create_opts = {
909 .name = "rbd-create-opts",
910 .head = QTAILQ_HEAD_INITIALIZER(qemu_rbd_create_opts.head),
911 .desc = {
913 .name = BLOCK_OPT_SIZE,
914 .type = QEMU_OPT_SIZE,
915 .help = "Virtual disk size"
918 .name = BLOCK_OPT_CLUSTER_SIZE,
919 .type = QEMU_OPT_SIZE,
920 .help = "RBD object size"
922 { /* end of list */ }
926 static BlockDriver bdrv_rbd = {
927 .format_name = "rbd",
928 .instance_size = sizeof(BDRVRBDState),
929 .bdrv_needs_filename = true,
930 .bdrv_file_open = qemu_rbd_open,
931 .bdrv_close = qemu_rbd_close,
932 .bdrv_create = qemu_rbd_create,
933 .bdrv_has_zero_init = bdrv_has_zero_init_1,
934 .bdrv_get_info = qemu_rbd_getinfo,
935 .create_opts = &qemu_rbd_create_opts,
936 .bdrv_getlength = qemu_rbd_getlength,
937 .bdrv_truncate = qemu_rbd_truncate,
938 .protocol_name = "rbd",
940 .bdrv_aio_readv = qemu_rbd_aio_readv,
941 .bdrv_aio_writev = qemu_rbd_aio_writev,
943 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
944 .bdrv_aio_flush = qemu_rbd_aio_flush,
945 #else
946 .bdrv_co_flush_to_disk = qemu_rbd_co_flush,
947 #endif
949 #ifdef LIBRBD_SUPPORTS_DISCARD
950 .bdrv_aio_discard = qemu_rbd_aio_discard,
951 #endif
953 .bdrv_snapshot_create = qemu_rbd_snap_create,
954 .bdrv_snapshot_delete = qemu_rbd_snap_remove,
955 .bdrv_snapshot_list = qemu_rbd_snap_list,
956 .bdrv_snapshot_goto = qemu_rbd_snap_rollback,
957 #ifdef LIBRBD_SUPPORTS_INVALIDATE
958 .bdrv_invalidate_cache = qemu_rbd_invalidate_cache,
959 #endif
962 static void bdrv_rbd_init(void)
964 bdrv_register(&bdrv_rbd);
967 block_init(bdrv_rbd_init);