cpus: properly inititalize CPU > 1 under single-threaded TCG
[qemu/ar7.git] / block / rbd.c
blobc7dd32e213b299c9110df691dbca8153286171ee
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 <rbd/librbd.h>
17 #include "qapi/error.h"
18 #include "qemu/error-report.h"
19 #include "qemu/option.h"
20 #include "block/block_int.h"
21 #include "crypto/secret.h"
22 #include "qemu/cutils.h"
23 #include "qapi/qmp/qstring.h"
24 #include "qapi/qmp/qdict.h"
25 #include "qapi/qmp/qjson.h"
26 #include "qapi/qmp/qlist.h"
29 * When specifying the image filename use:
31 * rbd:poolname/devicename[@snapshotname][:option1=value1[:option2=value2...]]
33 * poolname must be the name of an existing rados pool.
35 * devicename is the name of the rbd image.
37 * Each option given is used to configure rados, and may be any valid
38 * Ceph option, "id", or "conf".
40 * The "id" option indicates what user we should authenticate as to
41 * the Ceph cluster. If it is excluded we will use the Ceph default
42 * (normally 'admin').
44 * The "conf" option specifies a Ceph configuration file to read. If
45 * it is not specified, we will read from the default Ceph locations
46 * (e.g., /etc/ceph/ceph.conf). To avoid reading _any_ configuration
47 * file, specify conf=/dev/null.
49 * Configuration values containing :, @, or = can be escaped with a
50 * leading "\".
53 /* rbd_aio_discard added in 0.1.2 */
54 #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 2)
55 #define LIBRBD_SUPPORTS_DISCARD
56 #else
57 #undef LIBRBD_SUPPORTS_DISCARD
58 #endif
60 #define OBJ_MAX_SIZE (1UL << OBJ_DEFAULT_OBJ_ORDER)
62 #define RBD_MAX_SNAPS 100
64 /* The LIBRBD_SUPPORTS_IOVEC is defined in librbd.h */
65 #ifdef LIBRBD_SUPPORTS_IOVEC
66 #define LIBRBD_USE_IOVEC 1
67 #else
68 #define LIBRBD_USE_IOVEC 0
69 #endif
71 typedef enum {
72 RBD_AIO_READ,
73 RBD_AIO_WRITE,
74 RBD_AIO_DISCARD,
75 RBD_AIO_FLUSH
76 } RBDAIOCmd;
78 typedef struct RBDAIOCB {
79 BlockAIOCB common;
80 int64_t ret;
81 QEMUIOVector *qiov;
82 char *bounce;
83 RBDAIOCmd cmd;
84 int error;
85 struct BDRVRBDState *s;
86 } RBDAIOCB;
88 typedef struct RADOSCB {
89 RBDAIOCB *acb;
90 struct BDRVRBDState *s;
91 int64_t size;
92 char *buf;
93 int64_t ret;
94 } RADOSCB;
96 typedef struct BDRVRBDState {
97 rados_t cluster;
98 rados_ioctx_t io_ctx;
99 rbd_image_t image;
100 char *image_name;
101 char *snap;
102 } BDRVRBDState;
104 static char *qemu_rbd_next_tok(char *src, char delim, char **p)
106 char *end;
108 *p = NULL;
110 for (end = src; *end; ++end) {
111 if (*end == delim) {
112 break;
114 if (*end == '\\' && end[1] != '\0') {
115 end++;
118 if (*end == delim) {
119 *p = end + 1;
120 *end = '\0';
122 return src;
125 static void qemu_rbd_unescape(char *src)
127 char *p;
129 for (p = src; *src; ++src, ++p) {
130 if (*src == '\\' && src[1] != '\0') {
131 src++;
133 *p = *src;
135 *p = '\0';
138 static void qemu_rbd_parse_filename(const char *filename, QDict *options,
139 Error **errp)
141 const char *start;
142 char *p, *buf;
143 QList *keypairs = NULL;
144 char *found_str;
146 if (!strstart(filename, "rbd:", &start)) {
147 error_setg(errp, "File name must start with 'rbd:'");
148 return;
151 buf = g_strdup(start);
152 p = buf;
154 found_str = qemu_rbd_next_tok(p, '/', &p);
155 if (!p) {
156 error_setg(errp, "Pool name is required");
157 goto done;
159 qemu_rbd_unescape(found_str);
160 qdict_put_str(options, "pool", found_str);
162 if (strchr(p, '@')) {
163 found_str = qemu_rbd_next_tok(p, '@', &p);
164 qemu_rbd_unescape(found_str);
165 qdict_put_str(options, "image", found_str);
167 found_str = qemu_rbd_next_tok(p, ':', &p);
168 qemu_rbd_unescape(found_str);
169 qdict_put_str(options, "snapshot", found_str);
170 } else {
171 found_str = qemu_rbd_next_tok(p, ':', &p);
172 qemu_rbd_unescape(found_str);
173 qdict_put_str(options, "image", found_str);
175 if (!p) {
176 goto done;
179 /* The following are essentially all key/value pairs, and we treat
180 * 'id' and 'conf' a bit special. Key/value pairs may be in any order. */
181 while (p) {
182 char *name, *value;
183 name = qemu_rbd_next_tok(p, '=', &p);
184 if (!p) {
185 error_setg(errp, "conf option %s has no value", name);
186 break;
189 qemu_rbd_unescape(name);
191 value = qemu_rbd_next_tok(p, ':', &p);
192 qemu_rbd_unescape(value);
194 if (!strcmp(name, "conf")) {
195 qdict_put_str(options, "conf", value);
196 } else if (!strcmp(name, "id")) {
197 qdict_put_str(options, "user", value);
198 } else {
200 * We pass these internally to qemu_rbd_set_keypairs(), so
201 * we can get away with the simpler list of [ "key1",
202 * "value1", "key2", "value2" ] rather than a raw dict
203 * { "key1": "value1", "key2": "value2" } where we can't
204 * guarantee order, or even a more correct but complex
205 * [ { "key1": "value1" }, { "key2": "value2" } ]
207 if (!keypairs) {
208 keypairs = qlist_new();
210 qlist_append_str(keypairs, name);
211 qlist_append_str(keypairs, value);
215 if (keypairs) {
216 qdict_put(options, "=keyvalue-pairs",
217 qobject_to_json(QOBJECT(keypairs)));
220 done:
221 g_free(buf);
222 QDECREF(keypairs);
223 return;
227 static int qemu_rbd_set_auth(rados_t cluster, const char *secretid,
228 Error **errp)
230 if (secretid == 0) {
231 return 0;
234 gchar *secret = qcrypto_secret_lookup_as_base64(secretid,
235 errp);
236 if (!secret) {
237 return -1;
240 rados_conf_set(cluster, "key", secret);
241 g_free(secret);
243 return 0;
246 static int qemu_rbd_set_keypairs(rados_t cluster, const char *keypairs_json,
247 Error **errp)
249 QList *keypairs;
250 QString *name;
251 QString *value;
252 const char *key;
253 size_t remaining;
254 int ret = 0;
256 if (!keypairs_json) {
257 return ret;
259 keypairs = qobject_to_qlist(qobject_from_json(keypairs_json,
260 &error_abort));
261 remaining = qlist_size(keypairs) / 2;
262 assert(remaining);
264 while (remaining--) {
265 name = qobject_to_qstring(qlist_pop(keypairs));
266 value = qobject_to_qstring(qlist_pop(keypairs));
267 assert(name && value);
268 key = qstring_get_str(name);
270 ret = rados_conf_set(cluster, key, qstring_get_str(value));
271 QDECREF(name);
272 QDECREF(value);
273 if (ret < 0) {
274 error_setg_errno(errp, -ret, "invalid conf option %s", key);
275 ret = -EINVAL;
276 break;
280 QDECREF(keypairs);
281 return ret;
284 static void qemu_rbd_memset(RADOSCB *rcb, int64_t offs)
286 if (LIBRBD_USE_IOVEC) {
287 RBDAIOCB *acb = rcb->acb;
288 iov_memset(acb->qiov->iov, acb->qiov->niov, offs, 0,
289 acb->qiov->size - offs);
290 } else {
291 memset(rcb->buf + offs, 0, rcb->size - offs);
295 static QemuOptsList runtime_opts = {
296 .name = "rbd",
297 .head = QTAILQ_HEAD_INITIALIZER(runtime_opts.head),
298 .desc = {
300 .name = "pool",
301 .type = QEMU_OPT_STRING,
302 .help = "Rados pool name",
305 .name = "image",
306 .type = QEMU_OPT_STRING,
307 .help = "Image name in the pool",
310 .name = "conf",
311 .type = QEMU_OPT_STRING,
312 .help = "Rados config file location",
315 .name = "snapshot",
316 .type = QEMU_OPT_STRING,
317 .help = "Ceph snapshot name",
320 /* maps to 'id' in rados_create() */
321 .name = "user",
322 .type = QEMU_OPT_STRING,
323 .help = "Rados id name",
326 * server.* extracted manually, see qemu_rbd_mon_host()
329 .name = "password-secret",
330 .type = QEMU_OPT_STRING,
331 .help = "ID of secret providing the password",
335 * Keys for qemu_rbd_parse_filename(), not in the QAPI schema
339 * HACK: name starts with '=' so that qemu_opts_parse()
340 * can't set it
342 .name = "=keyvalue-pairs",
343 .type = QEMU_OPT_STRING,
344 .help = "Legacy rados key/value option parameters",
347 .name = "filename",
348 .type = QEMU_OPT_STRING,
350 { /* end of list */ }
354 static int coroutine_fn qemu_rbd_co_create_opts(const char *filename,
355 QemuOpts *opts,
356 Error **errp)
358 Error *local_err = NULL;
359 int64_t bytes = 0;
360 int64_t objsize;
361 int obj_order = 0;
362 const char *pool, *image_name, *conf, *user, *keypairs;
363 const char *secretid;
364 rados_t cluster;
365 rados_ioctx_t io_ctx;
366 QDict *options = NULL;
367 int ret = 0;
369 secretid = qemu_opt_get(opts, "password-secret");
371 /* Read out options */
372 bytes = ROUND_UP(qemu_opt_get_size_del(opts, BLOCK_OPT_SIZE, 0),
373 BDRV_SECTOR_SIZE);
374 objsize = qemu_opt_get_size_del(opts, BLOCK_OPT_CLUSTER_SIZE, 0);
375 if (objsize) {
376 if ((objsize - 1) & objsize) { /* not a power of 2? */
377 error_setg(errp, "obj size needs to be power of 2");
378 ret = -EINVAL;
379 goto exit;
381 if (objsize < 4096) {
382 error_setg(errp, "obj size too small");
383 ret = -EINVAL;
384 goto exit;
386 obj_order = ctz32(objsize);
389 options = qdict_new();
390 qemu_rbd_parse_filename(filename, options, &local_err);
391 if (local_err) {
392 ret = -EINVAL;
393 error_propagate(errp, local_err);
394 goto exit;
398 * Caution: while qdict_get_try_str() is fine, getting non-string
399 * types would require more care. When @options come from -blockdev
400 * or blockdev_add, its members are typed according to the QAPI
401 * schema, but when they come from -drive, they're all QString.
403 pool = qdict_get_try_str(options, "pool");
404 conf = qdict_get_try_str(options, "conf");
405 user = qdict_get_try_str(options, "user");
406 image_name = qdict_get_try_str(options, "image");
407 keypairs = qdict_get_try_str(options, "=keyvalue-pairs");
409 ret = rados_create(&cluster, user);
410 if (ret < 0) {
411 error_setg_errno(errp, -ret, "error initializing");
412 goto exit;
415 /* try default location when conf=NULL, but ignore failure */
416 ret = rados_conf_read_file(cluster, conf);
417 if (conf && ret < 0) {
418 error_setg_errno(errp, -ret, "error reading conf file %s", conf);
419 ret = -EIO;
420 goto shutdown;
423 ret = qemu_rbd_set_keypairs(cluster, keypairs, errp);
424 if (ret < 0) {
425 ret = -EIO;
426 goto shutdown;
429 if (qemu_rbd_set_auth(cluster, secretid, errp) < 0) {
430 ret = -EIO;
431 goto shutdown;
434 ret = rados_connect(cluster);
435 if (ret < 0) {
436 error_setg_errno(errp, -ret, "error connecting");
437 goto shutdown;
440 ret = rados_ioctx_create(cluster, pool, &io_ctx);
441 if (ret < 0) {
442 error_setg_errno(errp, -ret, "error opening pool %s", pool);
443 goto shutdown;
446 ret = rbd_create(io_ctx, image_name, bytes, &obj_order);
447 if (ret < 0) {
448 error_setg_errno(errp, -ret, "error rbd create");
451 rados_ioctx_destroy(io_ctx);
453 shutdown:
454 rados_shutdown(cluster);
456 exit:
457 QDECREF(options);
458 return ret;
462 * This aio completion is being called from rbd_finish_bh() and runs in qemu
463 * BH context.
465 static void qemu_rbd_complete_aio(RADOSCB *rcb)
467 RBDAIOCB *acb = rcb->acb;
468 int64_t r;
470 r = rcb->ret;
472 if (acb->cmd != RBD_AIO_READ) {
473 if (r < 0) {
474 acb->ret = r;
475 acb->error = 1;
476 } else if (!acb->error) {
477 acb->ret = rcb->size;
479 } else {
480 if (r < 0) {
481 qemu_rbd_memset(rcb, 0);
482 acb->ret = r;
483 acb->error = 1;
484 } else if (r < rcb->size) {
485 qemu_rbd_memset(rcb, r);
486 if (!acb->error) {
487 acb->ret = rcb->size;
489 } else if (!acb->error) {
490 acb->ret = r;
494 g_free(rcb);
496 if (!LIBRBD_USE_IOVEC) {
497 if (acb->cmd == RBD_AIO_READ) {
498 qemu_iovec_from_buf(acb->qiov, 0, acb->bounce, acb->qiov->size);
500 qemu_vfree(acb->bounce);
503 acb->common.cb(acb->common.opaque, (acb->ret > 0 ? 0 : acb->ret));
505 qemu_aio_unref(acb);
508 static char *qemu_rbd_mon_host(QDict *options, Error **errp)
510 const char **vals = g_new(const char *, qdict_size(options) + 1);
511 char keybuf[32];
512 const char *host, *port;
513 char *rados_str;
514 int i;
516 for (i = 0;; i++) {
517 sprintf(keybuf, "server.%d.host", i);
518 host = qdict_get_try_str(options, keybuf);
519 qdict_del(options, keybuf);
520 sprintf(keybuf, "server.%d.port", i);
521 port = qdict_get_try_str(options, keybuf);
522 qdict_del(options, keybuf);
523 if (!host && !port) {
524 break;
526 if (!host) {
527 error_setg(errp, "Parameter server.%d.host is missing", i);
528 rados_str = NULL;
529 goto out;
532 if (strchr(host, ':')) {
533 vals[i] = port ? g_strdup_printf("[%s]:%s", host, port)
534 : g_strdup_printf("[%s]", host);
535 } else {
536 vals[i] = port ? g_strdup_printf("%s:%s", host, port)
537 : g_strdup(host);
540 vals[i] = NULL;
542 rados_str = i ? g_strjoinv(";", (char **)vals) : NULL;
543 out:
544 g_strfreev((char **)vals);
545 return rados_str;
548 static int qemu_rbd_open(BlockDriverState *bs, QDict *options, int flags,
549 Error **errp)
551 BDRVRBDState *s = bs->opaque;
552 const char *pool, *snap, *conf, *user, *image_name, *keypairs;
553 const char *secretid, *filename;
554 QemuOpts *opts;
555 Error *local_err = NULL;
556 char *mon_host = NULL;
557 int r;
559 /* If we are given a filename, parse the filename, with precedence given to
560 * filename encoded options */
561 filename = qdict_get_try_str(options, "filename");
562 if (filename) {
563 warn_report("'filename' option specified. "
564 "This is an unsupported option, and may be deprecated "
565 "in the future");
566 qemu_rbd_parse_filename(filename, options, &local_err);
567 if (local_err) {
568 r = -EINVAL;
569 error_propagate(errp, local_err);
570 goto exit;
574 opts = qemu_opts_create(&runtime_opts, NULL, 0, &error_abort);
575 qemu_opts_absorb_qdict(opts, options, &local_err);
576 if (local_err) {
577 error_propagate(errp, local_err);
578 r = -EINVAL;
579 goto failed_opts;
582 mon_host = qemu_rbd_mon_host(options, &local_err);
583 if (local_err) {
584 error_propagate(errp, local_err);
585 r = -EINVAL;
586 goto failed_opts;
589 secretid = qemu_opt_get(opts, "password-secret");
591 pool = qemu_opt_get(opts, "pool");
592 conf = qemu_opt_get(opts, "conf");
593 snap = qemu_opt_get(opts, "snapshot");
594 user = qemu_opt_get(opts, "user");
595 image_name = qemu_opt_get(opts, "image");
596 keypairs = qemu_opt_get(opts, "=keyvalue-pairs");
598 if (!pool || !image_name) {
599 error_setg(errp, "Parameters 'pool' and 'image' are required");
600 r = -EINVAL;
601 goto failed_opts;
604 r = rados_create(&s->cluster, user);
605 if (r < 0) {
606 error_setg_errno(errp, -r, "error initializing");
607 goto failed_opts;
610 s->snap = g_strdup(snap);
611 s->image_name = g_strdup(image_name);
613 /* try default location when conf=NULL, but ignore failure */
614 r = rados_conf_read_file(s->cluster, conf);
615 if (conf && r < 0) {
616 error_setg_errno(errp, -r, "error reading conf file %s", conf);
617 goto failed_shutdown;
620 r = qemu_rbd_set_keypairs(s->cluster, keypairs, errp);
621 if (r < 0) {
622 goto failed_shutdown;
625 if (mon_host) {
626 r = rados_conf_set(s->cluster, "mon_host", mon_host);
627 if (r < 0) {
628 goto failed_shutdown;
632 if (qemu_rbd_set_auth(s->cluster, secretid, errp) < 0) {
633 r = -EIO;
634 goto failed_shutdown;
638 * Fallback to more conservative semantics if setting cache
639 * options fails. Ignore errors from setting rbd_cache because the
640 * only possible error is that the option does not exist, and
641 * librbd defaults to no caching. If write through caching cannot
642 * be set up, fall back to no caching.
644 if (flags & BDRV_O_NOCACHE) {
645 rados_conf_set(s->cluster, "rbd_cache", "false");
646 } else {
647 rados_conf_set(s->cluster, "rbd_cache", "true");
650 r = rados_connect(s->cluster);
651 if (r < 0) {
652 error_setg_errno(errp, -r, "error connecting");
653 goto failed_shutdown;
656 r = rados_ioctx_create(s->cluster, pool, &s->io_ctx);
657 if (r < 0) {
658 error_setg_errno(errp, -r, "error opening pool %s", pool);
659 goto failed_shutdown;
662 /* rbd_open is always r/w */
663 r = rbd_open(s->io_ctx, s->image_name, &s->image, s->snap);
664 if (r < 0) {
665 error_setg_errno(errp, -r, "error reading header from %s",
666 s->image_name);
667 goto failed_open;
670 /* If we are using an rbd snapshot, we must be r/o, otherwise
671 * leave as-is */
672 if (s->snap != NULL) {
673 if (!bdrv_is_read_only(bs)) {
674 error_report("Opening rbd snapshots without an explicit "
675 "read-only=on option is deprecated. Future versions "
676 "will refuse to open the image instead of "
677 "automatically marking the image read-only.");
678 r = bdrv_set_read_only(bs, true, &local_err);
679 if (r < 0) {
680 error_propagate(errp, local_err);
681 goto failed_open;
686 qemu_opts_del(opts);
687 return 0;
689 failed_open:
690 rados_ioctx_destroy(s->io_ctx);
691 failed_shutdown:
692 rados_shutdown(s->cluster);
693 g_free(s->snap);
694 g_free(s->image_name);
695 failed_opts:
696 qemu_opts_del(opts);
697 g_free(mon_host);
698 exit:
699 return r;
703 /* Since RBD is currently always opened R/W via the API,
704 * we just need to check if we are using a snapshot or not, in
705 * order to determine if we will allow it to be R/W */
706 static int qemu_rbd_reopen_prepare(BDRVReopenState *state,
707 BlockReopenQueue *queue, Error **errp)
709 BDRVRBDState *s = state->bs->opaque;
710 int ret = 0;
712 if (s->snap && state->flags & BDRV_O_RDWR) {
713 error_setg(errp,
714 "Cannot change node '%s' to r/w when using RBD snapshot",
715 bdrv_get_device_or_node_name(state->bs));
716 ret = -EINVAL;
719 return ret;
722 static void qemu_rbd_close(BlockDriverState *bs)
724 BDRVRBDState *s = bs->opaque;
726 rbd_close(s->image);
727 rados_ioctx_destroy(s->io_ctx);
728 g_free(s->snap);
729 g_free(s->image_name);
730 rados_shutdown(s->cluster);
733 static const AIOCBInfo rbd_aiocb_info = {
734 .aiocb_size = sizeof(RBDAIOCB),
737 static void rbd_finish_bh(void *opaque)
739 RADOSCB *rcb = opaque;
740 qemu_rbd_complete_aio(rcb);
744 * This is the callback function for rbd_aio_read and _write
746 * Note: this function is being called from a non qemu thread so
747 * we need to be careful about what we do here. Generally we only
748 * schedule a BH, and do the rest of the io completion handling
749 * from rbd_finish_bh() which runs in a qemu context.
751 static void rbd_finish_aiocb(rbd_completion_t c, RADOSCB *rcb)
753 RBDAIOCB *acb = rcb->acb;
755 rcb->ret = rbd_aio_get_return_value(c);
756 rbd_aio_release(c);
758 aio_bh_schedule_oneshot(bdrv_get_aio_context(acb->common.bs),
759 rbd_finish_bh, rcb);
762 static int rbd_aio_discard_wrapper(rbd_image_t image,
763 uint64_t off,
764 uint64_t len,
765 rbd_completion_t comp)
767 #ifdef LIBRBD_SUPPORTS_DISCARD
768 return rbd_aio_discard(image, off, len, comp);
769 #else
770 return -ENOTSUP;
771 #endif
774 static int rbd_aio_flush_wrapper(rbd_image_t image,
775 rbd_completion_t comp)
777 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
778 return rbd_aio_flush(image, comp);
779 #else
780 return -ENOTSUP;
781 #endif
784 static BlockAIOCB *rbd_start_aio(BlockDriverState *bs,
785 int64_t off,
786 QEMUIOVector *qiov,
787 int64_t size,
788 BlockCompletionFunc *cb,
789 void *opaque,
790 RBDAIOCmd cmd)
792 RBDAIOCB *acb;
793 RADOSCB *rcb = NULL;
794 rbd_completion_t c;
795 int r;
797 BDRVRBDState *s = bs->opaque;
799 acb = qemu_aio_get(&rbd_aiocb_info, bs, cb, opaque);
800 acb->cmd = cmd;
801 acb->qiov = qiov;
802 assert(!qiov || qiov->size == size);
804 rcb = g_new(RADOSCB, 1);
806 if (!LIBRBD_USE_IOVEC) {
807 if (cmd == RBD_AIO_DISCARD || cmd == RBD_AIO_FLUSH) {
808 acb->bounce = NULL;
809 } else {
810 acb->bounce = qemu_try_blockalign(bs, qiov->size);
811 if (acb->bounce == NULL) {
812 goto failed;
815 if (cmd == RBD_AIO_WRITE) {
816 qemu_iovec_to_buf(acb->qiov, 0, acb->bounce, qiov->size);
818 rcb->buf = acb->bounce;
821 acb->ret = 0;
822 acb->error = 0;
823 acb->s = s;
825 rcb->acb = acb;
826 rcb->s = acb->s;
827 rcb->size = size;
828 r = rbd_aio_create_completion(rcb, (rbd_callback_t) rbd_finish_aiocb, &c);
829 if (r < 0) {
830 goto failed;
833 switch (cmd) {
834 case RBD_AIO_WRITE:
835 #ifdef LIBRBD_SUPPORTS_IOVEC
836 r = rbd_aio_writev(s->image, qiov->iov, qiov->niov, off, c);
837 #else
838 r = rbd_aio_write(s->image, off, size, rcb->buf, c);
839 #endif
840 break;
841 case RBD_AIO_READ:
842 #ifdef LIBRBD_SUPPORTS_IOVEC
843 r = rbd_aio_readv(s->image, qiov->iov, qiov->niov, off, c);
844 #else
845 r = rbd_aio_read(s->image, off, size, rcb->buf, c);
846 #endif
847 break;
848 case RBD_AIO_DISCARD:
849 r = rbd_aio_discard_wrapper(s->image, off, size, c);
850 break;
851 case RBD_AIO_FLUSH:
852 r = rbd_aio_flush_wrapper(s->image, c);
853 break;
854 default:
855 r = -EINVAL;
858 if (r < 0) {
859 goto failed_completion;
861 return &acb->common;
863 failed_completion:
864 rbd_aio_release(c);
865 failed:
866 g_free(rcb);
867 if (!LIBRBD_USE_IOVEC) {
868 qemu_vfree(acb->bounce);
871 qemu_aio_unref(acb);
872 return NULL;
875 static BlockAIOCB *qemu_rbd_aio_readv(BlockDriverState *bs,
876 int64_t sector_num,
877 QEMUIOVector *qiov,
878 int nb_sectors,
879 BlockCompletionFunc *cb,
880 void *opaque)
882 return rbd_start_aio(bs, sector_num << BDRV_SECTOR_BITS, qiov,
883 (int64_t) nb_sectors << BDRV_SECTOR_BITS, cb, opaque,
884 RBD_AIO_READ);
887 static BlockAIOCB *qemu_rbd_aio_writev(BlockDriverState *bs,
888 int64_t sector_num,
889 QEMUIOVector *qiov,
890 int nb_sectors,
891 BlockCompletionFunc *cb,
892 void *opaque)
894 return rbd_start_aio(bs, sector_num << BDRV_SECTOR_BITS, qiov,
895 (int64_t) nb_sectors << BDRV_SECTOR_BITS, cb, opaque,
896 RBD_AIO_WRITE);
899 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
900 static BlockAIOCB *qemu_rbd_aio_flush(BlockDriverState *bs,
901 BlockCompletionFunc *cb,
902 void *opaque)
904 return rbd_start_aio(bs, 0, NULL, 0, cb, opaque, RBD_AIO_FLUSH);
907 #else
909 static int qemu_rbd_co_flush(BlockDriverState *bs)
911 #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 1)
912 /* rbd_flush added in 0.1.1 */
913 BDRVRBDState *s = bs->opaque;
914 return rbd_flush(s->image);
915 #else
916 return 0;
917 #endif
919 #endif
921 static int qemu_rbd_getinfo(BlockDriverState *bs, BlockDriverInfo *bdi)
923 BDRVRBDState *s = bs->opaque;
924 rbd_image_info_t info;
925 int r;
927 r = rbd_stat(s->image, &info, sizeof(info));
928 if (r < 0) {
929 return r;
932 bdi->cluster_size = info.obj_size;
933 return 0;
936 static int64_t qemu_rbd_getlength(BlockDriverState *bs)
938 BDRVRBDState *s = bs->opaque;
939 rbd_image_info_t info;
940 int r;
942 r = rbd_stat(s->image, &info, sizeof(info));
943 if (r < 0) {
944 return r;
947 return info.size;
950 static int qemu_rbd_truncate(BlockDriverState *bs, int64_t offset,
951 PreallocMode prealloc, Error **errp)
953 BDRVRBDState *s = bs->opaque;
954 int r;
956 if (prealloc != PREALLOC_MODE_OFF) {
957 error_setg(errp, "Unsupported preallocation mode '%s'",
958 PreallocMode_str(prealloc));
959 return -ENOTSUP;
962 r = rbd_resize(s->image, offset);
963 if (r < 0) {
964 error_setg_errno(errp, -r, "Failed to resize file");
965 return r;
968 return 0;
971 static int qemu_rbd_snap_create(BlockDriverState *bs,
972 QEMUSnapshotInfo *sn_info)
974 BDRVRBDState *s = bs->opaque;
975 int r;
977 if (sn_info->name[0] == '\0') {
978 return -EINVAL; /* we need a name for rbd snapshots */
982 * rbd snapshots are using the name as the user controlled unique identifier
983 * we can't use the rbd snapid for that purpose, as it can't be set
985 if (sn_info->id_str[0] != '\0' &&
986 strcmp(sn_info->id_str, sn_info->name) != 0) {
987 return -EINVAL;
990 if (strlen(sn_info->name) >= sizeof(sn_info->id_str)) {
991 return -ERANGE;
994 r = rbd_snap_create(s->image, sn_info->name);
995 if (r < 0) {
996 error_report("failed to create snap: %s", strerror(-r));
997 return r;
1000 return 0;
1003 static int qemu_rbd_snap_remove(BlockDriverState *bs,
1004 const char *snapshot_id,
1005 const char *snapshot_name,
1006 Error **errp)
1008 BDRVRBDState *s = bs->opaque;
1009 int r;
1011 if (!snapshot_name) {
1012 error_setg(errp, "rbd need a valid snapshot name");
1013 return -EINVAL;
1016 /* If snapshot_id is specified, it must be equal to name, see
1017 qemu_rbd_snap_list() */
1018 if (snapshot_id && strcmp(snapshot_id, snapshot_name)) {
1019 error_setg(errp,
1020 "rbd do not support snapshot id, it should be NULL or "
1021 "equal to snapshot name");
1022 return -EINVAL;
1025 r = rbd_snap_remove(s->image, snapshot_name);
1026 if (r < 0) {
1027 error_setg_errno(errp, -r, "Failed to remove the snapshot");
1029 return r;
1032 static int qemu_rbd_snap_rollback(BlockDriverState *bs,
1033 const char *snapshot_name)
1035 BDRVRBDState *s = bs->opaque;
1037 return rbd_snap_rollback(s->image, snapshot_name);
1040 static int qemu_rbd_snap_list(BlockDriverState *bs,
1041 QEMUSnapshotInfo **psn_tab)
1043 BDRVRBDState *s = bs->opaque;
1044 QEMUSnapshotInfo *sn_info, *sn_tab = NULL;
1045 int i, snap_count;
1046 rbd_snap_info_t *snaps;
1047 int max_snaps = RBD_MAX_SNAPS;
1049 do {
1050 snaps = g_new(rbd_snap_info_t, max_snaps);
1051 snap_count = rbd_snap_list(s->image, snaps, &max_snaps);
1052 if (snap_count <= 0) {
1053 g_free(snaps);
1055 } while (snap_count == -ERANGE);
1057 if (snap_count <= 0) {
1058 goto done;
1061 sn_tab = g_new0(QEMUSnapshotInfo, snap_count);
1063 for (i = 0; i < snap_count; i++) {
1064 const char *snap_name = snaps[i].name;
1066 sn_info = sn_tab + i;
1067 pstrcpy(sn_info->id_str, sizeof(sn_info->id_str), snap_name);
1068 pstrcpy(sn_info->name, sizeof(sn_info->name), snap_name);
1070 sn_info->vm_state_size = snaps[i].size;
1071 sn_info->date_sec = 0;
1072 sn_info->date_nsec = 0;
1073 sn_info->vm_clock_nsec = 0;
1075 rbd_snap_list_end(snaps);
1076 g_free(snaps);
1078 done:
1079 *psn_tab = sn_tab;
1080 return snap_count;
1083 #ifdef LIBRBD_SUPPORTS_DISCARD
1084 static BlockAIOCB *qemu_rbd_aio_pdiscard(BlockDriverState *bs,
1085 int64_t offset,
1086 int bytes,
1087 BlockCompletionFunc *cb,
1088 void *opaque)
1090 return rbd_start_aio(bs, offset, NULL, bytes, cb, opaque,
1091 RBD_AIO_DISCARD);
1093 #endif
1095 #ifdef LIBRBD_SUPPORTS_INVALIDATE
1096 static void qemu_rbd_invalidate_cache(BlockDriverState *bs,
1097 Error **errp)
1099 BDRVRBDState *s = bs->opaque;
1100 int r = rbd_invalidate_cache(s->image);
1101 if (r < 0) {
1102 error_setg_errno(errp, -r, "Failed to invalidate the cache");
1105 #endif
1107 static QemuOptsList qemu_rbd_create_opts = {
1108 .name = "rbd-create-opts",
1109 .head = QTAILQ_HEAD_INITIALIZER(qemu_rbd_create_opts.head),
1110 .desc = {
1112 .name = BLOCK_OPT_SIZE,
1113 .type = QEMU_OPT_SIZE,
1114 .help = "Virtual disk size"
1117 .name = BLOCK_OPT_CLUSTER_SIZE,
1118 .type = QEMU_OPT_SIZE,
1119 .help = "RBD object size"
1122 .name = "password-secret",
1123 .type = QEMU_OPT_STRING,
1124 .help = "ID of secret providing the password",
1126 { /* end of list */ }
1130 static BlockDriver bdrv_rbd = {
1131 .format_name = "rbd",
1132 .instance_size = sizeof(BDRVRBDState),
1133 .bdrv_parse_filename = qemu_rbd_parse_filename,
1134 .bdrv_file_open = qemu_rbd_open,
1135 .bdrv_close = qemu_rbd_close,
1136 .bdrv_reopen_prepare = qemu_rbd_reopen_prepare,
1137 .bdrv_co_create_opts = qemu_rbd_co_create_opts,
1138 .bdrv_has_zero_init = bdrv_has_zero_init_1,
1139 .bdrv_get_info = qemu_rbd_getinfo,
1140 .create_opts = &qemu_rbd_create_opts,
1141 .bdrv_getlength = qemu_rbd_getlength,
1142 .bdrv_truncate = qemu_rbd_truncate,
1143 .protocol_name = "rbd",
1145 .bdrv_aio_readv = qemu_rbd_aio_readv,
1146 .bdrv_aio_writev = qemu_rbd_aio_writev,
1148 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
1149 .bdrv_aio_flush = qemu_rbd_aio_flush,
1150 #else
1151 .bdrv_co_flush_to_disk = qemu_rbd_co_flush,
1152 #endif
1154 #ifdef LIBRBD_SUPPORTS_DISCARD
1155 .bdrv_aio_pdiscard = qemu_rbd_aio_pdiscard,
1156 #endif
1158 .bdrv_snapshot_create = qemu_rbd_snap_create,
1159 .bdrv_snapshot_delete = qemu_rbd_snap_remove,
1160 .bdrv_snapshot_list = qemu_rbd_snap_list,
1161 .bdrv_snapshot_goto = qemu_rbd_snap_rollback,
1162 #ifdef LIBRBD_SUPPORTS_INVALIDATE
1163 .bdrv_invalidate_cache = qemu_rbd_invalidate_cache,
1164 #endif
1167 static void bdrv_rbd_init(void)
1169 bdrv_register(&bdrv_rbd);
1172 block_init(bdrv_rbd_init);