hw/elf_ops: Fix a typo
[qemu/ar7.git] / block / rbd.c
blob9071a00e3f757daa10cad6d0c9f9da0b85ebd8c5
1 /*
2  * QEMU Block driver for RADOS (Ceph)
3  *
4  * Copyright (C) 2010-2011 Christian Brunner <chb@muc.de>,
5  *                         Josh Durgin <josh.durgin@dreamhost.com>
6  *
7  * This work is licensed under the terms of the GNU GPL, version 2.  See
8  * the COPYING file in the top-level directory.
9  *
10  * Contributions after 2012-01-13 are licensed under the terms of the
11  * GNU GPL, version 2 or (at your option) any later version.
12  */
14 #include "qemu/osdep.h"
16 #include <rbd/librbd.h>
17 #include "qapi/error.h"
18 #include "qemu/error-report.h"
19 #include "qemu/module.h"
20 #include "qemu/option.h"
21 #include "block/block_int.h"
22 #include "block/qdict.h"
23 #include "crypto/secret.h"
24 #include "qemu/cutils.h"
25 #include "sysemu/replay.h"
26 #include "qapi/qmp/qstring.h"
27 #include "qapi/qmp/qdict.h"
28 #include "qapi/qmp/qjson.h"
29 #include "qapi/qmp/qlist.h"
30 #include "qapi/qobject-input-visitor.h"
31 #include "qapi/qapi-visit-block-core.h"
34  * When specifying the image filename use:
35  *
36  * rbd:poolname/devicename[@snapshotname][:option1=value1[:option2=value2...]]
37  *
38  * poolname must be the name of an existing rados pool.
39  *
40  * devicename is the name of the rbd image.
41  *
42  * Each option given is used to configure rados, and may be any valid
43  * Ceph option, "id", or "conf".
44  *
45  * The "id" option indicates what user we should authenticate as to
46  * the Ceph cluster.  If it is excluded we will use the Ceph default
47  * (normally 'admin').
48  *
49  * The "conf" option specifies a Ceph configuration file to read.  If
50  * it is not specified, we will read from the default Ceph locations
51  * (e.g., /etc/ceph/ceph.conf).  To avoid reading _any_ configuration
52  * file, specify conf=/dev/null.
53  *
54  * Configuration values containing :, @, or = can be escaped with a
55  * leading "\".
56  */
58 /* rbd_aio_discard added in 0.1.2 */
59 #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 2)
60 #define LIBRBD_SUPPORTS_DISCARD
61 #else
62 #undef LIBRBD_SUPPORTS_DISCARD
63 #endif
65 #define OBJ_MAX_SIZE (1UL << OBJ_DEFAULT_OBJ_ORDER)
67 #define RBD_MAX_SNAPS 100
69 /* The LIBRBD_SUPPORTS_IOVEC is defined in librbd.h */
70 #ifdef LIBRBD_SUPPORTS_IOVEC
71 #define LIBRBD_USE_IOVEC 1
72 #else
73 #define LIBRBD_USE_IOVEC 0
74 #endif
76 typedef enum {
77     RBD_AIO_READ,
78     RBD_AIO_WRITE,
79     RBD_AIO_DISCARD,
80     RBD_AIO_FLUSH
81 } RBDAIOCmd;
83 typedef struct RBDAIOCB {
84     BlockAIOCB common;
85     int64_t ret;
86     QEMUIOVector *qiov;
87     char *bounce;
88     RBDAIOCmd cmd;
89     int error;
90     struct BDRVRBDState *s;
91 } RBDAIOCB;
93 typedef struct RADOSCB {
94     RBDAIOCB *acb;
95     struct BDRVRBDState *s;
96     int64_t size;
97     char *buf;
98     int64_t ret;
99 } RADOSCB;
101 typedef struct BDRVRBDState {
102     rados_t cluster;
103     rados_ioctx_t io_ctx;
104     rbd_image_t image;
105     char *image_name;
106     char *snap;
107     char *namespace;
108     uint64_t image_size;
109 } BDRVRBDState;
111 static int qemu_rbd_connect(rados_t *cluster, rados_ioctx_t *io_ctx,
112                             BlockdevOptionsRbd *opts, bool cache,
113                             const char *keypairs, const char *secretid,
114                             Error **errp);
116 static char *qemu_rbd_next_tok(char *src, char delim, char **p)
118     char *end;
120     *p = NULL;
122     for (end = src; *end; ++end) {
123         if (*end == delim) {
124             break;
125         }
126         if (*end == '\\' && end[1] != '\0') {
127             end++;
128         }
129     }
130     if (*end == delim) {
131         *p = end + 1;
132         *end = '\0';
133     }
134     return src;
137 static void qemu_rbd_unescape(char *src)
139     char *p;
141     for (p = src; *src; ++src, ++p) {
142         if (*src == '\\' && src[1] != '\0') {
143             src++;
144         }
145         *p = *src;
146     }
147     *p = '\0';
150 static void qemu_rbd_parse_filename(const char *filename, QDict *options,
151                                     Error **errp)
153     const char *start;
154     char *p, *buf;
155     QList *keypairs = NULL;
156     char *found_str, *image_name;
158     if (!strstart(filename, "rbd:", &start)) {
159         error_setg(errp, "File name must start with 'rbd:'");
160         return;
161     }
163     buf = g_strdup(start);
164     p = buf;
166     found_str = qemu_rbd_next_tok(p, '/', &p);
167     if (!p) {
168         error_setg(errp, "Pool name is required");
169         goto done;
170     }
171     qemu_rbd_unescape(found_str);
172     qdict_put_str(options, "pool", found_str);
174     if (strchr(p, '@')) {
175         image_name = qemu_rbd_next_tok(p, '@', &p);
177         found_str = qemu_rbd_next_tok(p, ':', &p);
178         qemu_rbd_unescape(found_str);
179         qdict_put_str(options, "snapshot", found_str);
180     } else {
181         image_name = qemu_rbd_next_tok(p, ':', &p);
182     }
183     /* Check for namespace in the image_name */
184     if (strchr(image_name, '/')) {
185         found_str = qemu_rbd_next_tok(image_name, '/', &image_name);
186         qemu_rbd_unescape(found_str);
187         qdict_put_str(options, "namespace", found_str);
188     } else {
189         qdict_put_str(options, "namespace", "");
190     }
191     qemu_rbd_unescape(image_name);
192     qdict_put_str(options, "image", image_name);
193     if (!p) {
194         goto done;
195     }
197     /* The following are essentially all key/value pairs, and we treat
198      * 'id' and 'conf' a bit special.  Key/value pairs may be in any order. */
199     while (p) {
200         char *name, *value;
201         name = qemu_rbd_next_tok(p, '=', &p);
202         if (!p) {
203             error_setg(errp, "conf option %s has no value", name);
204             break;
205         }
207         qemu_rbd_unescape(name);
209         value = qemu_rbd_next_tok(p, ':', &p);
210         qemu_rbd_unescape(value);
212         if (!strcmp(name, "conf")) {
213             qdict_put_str(options, "conf", value);
214         } else if (!strcmp(name, "id")) {
215             qdict_put_str(options, "user", value);
216         } else {
217             /*
218              * We pass these internally to qemu_rbd_set_keypairs(), so
219              * we can get away with the simpler list of [ "key1",
220              * "value1", "key2", "value2" ] rather than a raw dict
221              * { "key1": "value1", "key2": "value2" } where we can't
222              * guarantee order, or even a more correct but complex
223              * [ { "key1": "value1" }, { "key2": "value2" } ]
224              */
225             if (!keypairs) {
226                 keypairs = qlist_new();
227             }
228             qlist_append_str(keypairs, name);
229             qlist_append_str(keypairs, value);
230         }
231     }
233     if (keypairs) {
234         qdict_put(options, "=keyvalue-pairs",
235                   qstring_from_gstring(qobject_to_json(QOBJECT(keypairs))));
236     }
238 done:
239     g_free(buf);
240     qobject_unref(keypairs);
241     return;
245 static void qemu_rbd_refresh_limits(BlockDriverState *bs, Error **errp)
247     /* XXX Does RBD support AIO on less than 512-byte alignment? */
248     bs->bl.request_alignment = 512;
252 static int qemu_rbd_set_auth(rados_t cluster, BlockdevOptionsRbd *opts,
253                              Error **errp)
255     char *key, *acr;
256     int r;
257     GString *accu;
258     RbdAuthModeList *auth;
260     if (opts->key_secret) {
261         key = qcrypto_secret_lookup_as_base64(opts->key_secret, errp);
262         if (!key) {
263             return -EIO;
264         }
265         r = rados_conf_set(cluster, "key", key);
266         g_free(key);
267         if (r < 0) {
268             error_setg_errno(errp, -r, "Could not set 'key'");
269             return r;
270         }
271     }
273     if (opts->has_auth_client_required) {
274         accu = g_string_new("");
275         for (auth = opts->auth_client_required; auth; auth = auth->next) {
276             if (accu->str[0]) {
277                 g_string_append_c(accu, ';');
278             }
279             g_string_append(accu, RbdAuthMode_str(auth->value));
280         }
281         acr = g_string_free(accu, FALSE);
282         r = rados_conf_set(cluster, "auth_client_required", acr);
283         g_free(acr);
284         if (r < 0) {
285             error_setg_errno(errp, -r,
286                              "Could not set 'auth_client_required'");
287             return r;
288         }
289     }
291     return 0;
294 static int qemu_rbd_set_keypairs(rados_t cluster, const char *keypairs_json,
295                                  Error **errp)
297     QList *keypairs;
298     QString *name;
299     QString *value;
300     const char *key;
301     size_t remaining;
302     int ret = 0;
304     if (!keypairs_json) {
305         return ret;
306     }
307     keypairs = qobject_to(QList,
308                           qobject_from_json(keypairs_json, &error_abort));
309     remaining = qlist_size(keypairs) / 2;
310     assert(remaining);
312     while (remaining--) {
313         name = qobject_to(QString, qlist_pop(keypairs));
314         value = qobject_to(QString, qlist_pop(keypairs));
315         assert(name && value);
316         key = qstring_get_str(name);
318         ret = rados_conf_set(cluster, key, qstring_get_str(value));
319         qobject_unref(value);
320         if (ret < 0) {
321             error_setg_errno(errp, -ret, "invalid conf option %s", key);
322             qobject_unref(name);
323             ret = -EINVAL;
324             break;
325         }
326         qobject_unref(name);
327     }
329     qobject_unref(keypairs);
330     return ret;
333 static void qemu_rbd_memset(RADOSCB *rcb, int64_t offs)
335     if (LIBRBD_USE_IOVEC) {
336         RBDAIOCB *acb = rcb->acb;
337         iov_memset(acb->qiov->iov, acb->qiov->niov, offs, 0,
338                    acb->qiov->size - offs);
339     } else {
340         memset(rcb->buf + offs, 0, rcb->size - offs);
341     }
344 /* FIXME Deprecate and remove keypairs or make it available in QMP. */
345 static int qemu_rbd_do_create(BlockdevCreateOptions *options,
346                               const char *keypairs, const char *password_secret,
347                               Error **errp)
349     BlockdevCreateOptionsRbd *opts = &options->u.rbd;
350     rados_t cluster;
351     rados_ioctx_t io_ctx;
352     int obj_order = 0;
353     int ret;
355     assert(options->driver == BLOCKDEV_DRIVER_RBD);
356     if (opts->location->has_snapshot) {
357         error_setg(errp, "Can't use snapshot name for image creation");
358         return -EINVAL;
359     }
361     if (opts->has_cluster_size) {
362         int64_t objsize = opts->cluster_size;
363         if ((objsize - 1) & objsize) {    /* not a power of 2? */
364             error_setg(errp, "obj size needs to be power of 2");
365             return -EINVAL;
366         }
367         if (objsize < 4096) {
368             error_setg(errp, "obj size too small");
369             return -EINVAL;
370         }
371         obj_order = ctz32(objsize);
372     }
374     ret = qemu_rbd_connect(&cluster, &io_ctx, opts->location, false, keypairs,
375                            password_secret, errp);
376     if (ret < 0) {
377         return ret;
378     }
380     ret = rbd_create(io_ctx, opts->location->image, opts->size, &obj_order);
381     if (ret < 0) {
382         error_setg_errno(errp, -ret, "error rbd create");
383         goto out;
384     }
386     ret = 0;
387 out:
388     rados_ioctx_destroy(io_ctx);
389     rados_shutdown(cluster);
390     return ret;
393 static int qemu_rbd_co_create(BlockdevCreateOptions *options, Error **errp)
395     return qemu_rbd_do_create(options, NULL, NULL, errp);
398 static int coroutine_fn qemu_rbd_co_create_opts(BlockDriver *drv,
399                                                 const char *filename,
400                                                 QemuOpts *opts,
401                                                 Error **errp)
403     BlockdevCreateOptions *create_options;
404     BlockdevCreateOptionsRbd *rbd_opts;
405     BlockdevOptionsRbd *loc;
406     Error *local_err = NULL;
407     const char *keypairs, *password_secret;
408     QDict *options = NULL;
409     int ret = 0;
411     create_options = g_new0(BlockdevCreateOptions, 1);
412     create_options->driver = BLOCKDEV_DRIVER_RBD;
413     rbd_opts = &create_options->u.rbd;
415     rbd_opts->location = g_new0(BlockdevOptionsRbd, 1);
417     password_secret = qemu_opt_get(opts, "password-secret");
419     /* Read out options */
420     rbd_opts->size = ROUND_UP(qemu_opt_get_size_del(opts, BLOCK_OPT_SIZE, 0),
421                               BDRV_SECTOR_SIZE);
422     rbd_opts->cluster_size = qemu_opt_get_size_del(opts,
423                                                    BLOCK_OPT_CLUSTER_SIZE, 0);
424     rbd_opts->has_cluster_size = (rbd_opts->cluster_size != 0);
426     options = qdict_new();
427     qemu_rbd_parse_filename(filename, options, &local_err);
428     if (local_err) {
429         ret = -EINVAL;
430         error_propagate(errp, local_err);
431         goto exit;
432     }
434     /*
435      * Caution: while qdict_get_try_str() is fine, getting non-string
436      * types would require more care.  When @options come from -blockdev
437      * or blockdev_add, its members are typed according to the QAPI
438      * schema, but when they come from -drive, they're all QString.
439      */
440     loc = rbd_opts->location;
441     loc->pool        = g_strdup(qdict_get_try_str(options, "pool"));
442     loc->conf        = g_strdup(qdict_get_try_str(options, "conf"));
443     loc->has_conf    = !!loc->conf;
444     loc->user        = g_strdup(qdict_get_try_str(options, "user"));
445     loc->has_user    = !!loc->user;
446     loc->q_namespace = g_strdup(qdict_get_try_str(options, "namespace"));
447     loc->image       = g_strdup(qdict_get_try_str(options, "image"));
448     keypairs         = qdict_get_try_str(options, "=keyvalue-pairs");
450     ret = qemu_rbd_do_create(create_options, keypairs, password_secret, errp);
451     if (ret < 0) {
452         goto exit;
453     }
455 exit:
456     qobject_unref(options);
457     qapi_free_BlockdevCreateOptions(create_options);
458     return ret;
462  * This aio completion is being called from rbd_finish_bh() and runs in qemu
463  * BH context.
464  */
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;
478         }
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;
488             }
489         } else if (!acb->error) {
490             acb->ret = r;
491         }
492     }
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);
499         }
500         qemu_vfree(acb->bounce);
501     }
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(BlockdevOptionsRbd *opts, Error **errp)
510     const char **vals;
511     const char *host, *port;
512     char *rados_str;
513     InetSocketAddressBaseList *p;
514     int i, cnt;
516     if (!opts->has_server) {
517         return NULL;
518     }
520     for (cnt = 0, p = opts->server; p; p = p->next) {
521         cnt++;
522     }
524     vals = g_new(const char *, cnt + 1);
526     for (i = 0, p = opts->server; p; p = p->next, i++) {
527         host = p->value->host;
528         port = p->value->port;
530         if (strchr(host, ':')) {
531             vals[i] = g_strdup_printf("[%s]:%s", host, port);
532         } else {
533             vals[i] = g_strdup_printf("%s:%s", host, port);
534         }
535     }
536     vals[i] = NULL;
538     rados_str = i ? g_strjoinv(";", (char **)vals) : NULL;
539     g_strfreev((char **)vals);
540     return rados_str;
543 static int qemu_rbd_connect(rados_t *cluster, rados_ioctx_t *io_ctx,
544                             BlockdevOptionsRbd *opts, bool cache,
545                             const char *keypairs, const char *secretid,
546                             Error **errp)
548     char *mon_host = NULL;
549     Error *local_err = NULL;
550     int r;
552     if (secretid) {
553         if (opts->key_secret) {
554             error_setg(errp,
555                        "Legacy 'password-secret' clashes with 'key-secret'");
556             return -EINVAL;
557         }
558         opts->key_secret = g_strdup(secretid);
559         opts->has_key_secret = true;
560     }
562     mon_host = qemu_rbd_mon_host(opts, &local_err);
563     if (local_err) {
564         error_propagate(errp, local_err);
565         r = -EINVAL;
566         goto failed_opts;
567     }
569     r = rados_create(cluster, opts->user);
570     if (r < 0) {
571         error_setg_errno(errp, -r, "error initializing");
572         goto failed_opts;
573     }
575     /* try default location when conf=NULL, but ignore failure */
576     r = rados_conf_read_file(*cluster, opts->conf);
577     if (opts->has_conf && r < 0) {
578         error_setg_errno(errp, -r, "error reading conf file %s", opts->conf);
579         goto failed_shutdown;
580     }
582     r = qemu_rbd_set_keypairs(*cluster, keypairs, errp);
583     if (r < 0) {
584         goto failed_shutdown;
585     }
587     if (mon_host) {
588         r = rados_conf_set(*cluster, "mon_host", mon_host);
589         if (r < 0) {
590             goto failed_shutdown;
591         }
592     }
594     r = qemu_rbd_set_auth(*cluster, opts, errp);
595     if (r < 0) {
596         goto failed_shutdown;
597     }
599     /*
600      * Fallback to more conservative semantics if setting cache
601      * options fails. Ignore errors from setting rbd_cache because the
602      * only possible error is that the option does not exist, and
603      * librbd defaults to no caching. If write through caching cannot
604      * be set up, fall back to no caching.
605      */
606     if (cache) {
607         rados_conf_set(*cluster, "rbd_cache", "true");
608     } else {
609         rados_conf_set(*cluster, "rbd_cache", "false");
610     }
612     r = rados_connect(*cluster);
613     if (r < 0) {
614         error_setg_errno(errp, -r, "error connecting");
615         goto failed_shutdown;
616     }
618     r = rados_ioctx_create(*cluster, opts->pool, io_ctx);
619     if (r < 0) {
620         error_setg_errno(errp, -r, "error opening pool %s", opts->pool);
621         goto failed_shutdown;
622     }
623     /*
624      * Set the namespace after opening the io context on the pool,
625      * if nspace == NULL or if nspace == "", it is just as we did nothing
626      */
627     rados_ioctx_set_namespace(*io_ctx, opts->q_namespace);
629     return 0;
631 failed_shutdown:
632     rados_shutdown(*cluster);
633 failed_opts:
634     g_free(mon_host);
635     return r;
638 static int qemu_rbd_convert_options(QDict *options, BlockdevOptionsRbd **opts,
639                                     Error **errp)
641     Visitor *v;
643     /* Convert the remaining options into a QAPI object */
644     v = qobject_input_visitor_new_flat_confused(options, errp);
645     if (!v) {
646         return -EINVAL;
647     }
649     visit_type_BlockdevOptionsRbd(v, NULL, opts, errp);
650     visit_free(v);
651     if (!opts) {
652         return -EINVAL;
653     }
655     return 0;
658 static int qemu_rbd_attempt_legacy_options(QDict *options,
659                                            BlockdevOptionsRbd **opts,
660                                            char **keypairs)
662     char *filename;
663     int r;
665     filename = g_strdup(qdict_get_try_str(options, "filename"));
666     if (!filename) {
667         return -EINVAL;
668     }
669     qdict_del(options, "filename");
671     qemu_rbd_parse_filename(filename, options, NULL);
673     /* keypairs freed by caller */
674     *keypairs = g_strdup(qdict_get_try_str(options, "=keyvalue-pairs"));
675     if (*keypairs) {
676         qdict_del(options, "=keyvalue-pairs");
677     }
679     r = qemu_rbd_convert_options(options, opts, NULL);
681     g_free(filename);
682     return r;
685 static int qemu_rbd_open(BlockDriverState *bs, QDict *options, int flags,
686                          Error **errp)
688     BDRVRBDState *s = bs->opaque;
689     BlockdevOptionsRbd *opts = NULL;
690     const QDictEntry *e;
691     Error *local_err = NULL;
692     char *keypairs, *secretid;
693     int r;
695     keypairs = g_strdup(qdict_get_try_str(options, "=keyvalue-pairs"));
696     if (keypairs) {
697         qdict_del(options, "=keyvalue-pairs");
698     }
700     secretid = g_strdup(qdict_get_try_str(options, "password-secret"));
701     if (secretid) {
702         qdict_del(options, "password-secret");
703     }
705     r = qemu_rbd_convert_options(options, &opts, &local_err);
706     if (local_err) {
707         /* If keypairs are present, that means some options are present in
708          * the modern option format.  Don't attempt to parse legacy option
709          * formats, as we won't support mixed usage. */
710         if (keypairs) {
711             error_propagate(errp, local_err);
712             goto out;
713         }
715         /* If the initial attempt to convert and process the options failed,
716          * we may be attempting to open an image file that has the rbd options
717          * specified in the older format consisting of all key/value pairs
718          * encoded in the filename.  Go ahead and attempt to parse the
719          * filename, and see if we can pull out the required options. */
720         r = qemu_rbd_attempt_legacy_options(options, &opts, &keypairs);
721         if (r < 0) {
722             /* Propagate the original error, not the legacy parsing fallback
723              * error, as the latter was just a best-effort attempt. */
724             error_propagate(errp, local_err);
725             goto out;
726         }
727         /* Take care whenever deciding to actually deprecate; once this ability
728          * is removed, we will not be able to open any images with legacy-styled
729          * backing image strings. */
730         warn_report("RBD options encoded in the filename as keyvalue pairs "
731                     "is deprecated");
732     }
734     /* Remove the processed options from the QDict (the visitor processes
735      * _all_ options in the QDict) */
736     while ((e = qdict_first(options))) {
737         qdict_del(options, e->key);
738     }
740     r = qemu_rbd_connect(&s->cluster, &s->io_ctx, opts,
741                          !(flags & BDRV_O_NOCACHE), keypairs, secretid, errp);
742     if (r < 0) {
743         goto out;
744     }
746     s->snap = g_strdup(opts->snapshot);
747     s->image_name = g_strdup(opts->image);
749     /* rbd_open is always r/w */
750     r = rbd_open(s->io_ctx, s->image_name, &s->image, s->snap);
751     if (r < 0) {
752         error_setg_errno(errp, -r, "error reading header from %s",
753                          s->image_name);
754         goto failed_open;
755     }
757     r = rbd_get_size(s->image, &s->image_size);
758     if (r < 0) {
759         error_setg_errno(errp, -r, "error getting image size from %s",
760                          s->image_name);
761         rbd_close(s->image);
762         goto failed_open;
763     }
765     /* If we are using an rbd snapshot, we must be r/o, otherwise
766      * leave as-is */
767     if (s->snap != NULL) {
768         r = bdrv_apply_auto_read_only(bs, "rbd snapshots are read-only", errp);
769         if (r < 0) {
770             rbd_close(s->image);
771             goto failed_open;
772         }
773     }
775     /* When extending regular files, we get zeros from the OS */
776     bs->supported_truncate_flags = BDRV_REQ_ZERO_WRITE;
778     r = 0;
779     goto out;
781 failed_open:
782     rados_ioctx_destroy(s->io_ctx);
783     g_free(s->snap);
784     g_free(s->image_name);
785     rados_shutdown(s->cluster);
786 out:
787     qapi_free_BlockdevOptionsRbd(opts);
788     g_free(keypairs);
789     g_free(secretid);
790     return r;
794 /* Since RBD is currently always opened R/W via the API,
795  * we just need to check if we are using a snapshot or not, in
796  * order to determine if we will allow it to be R/W */
797 static int qemu_rbd_reopen_prepare(BDRVReopenState *state,
798                                    BlockReopenQueue *queue, Error **errp)
800     BDRVRBDState *s = state->bs->opaque;
801     int ret = 0;
803     if (s->snap && state->flags & BDRV_O_RDWR) {
804         error_setg(errp,
805                    "Cannot change node '%s' to r/w when using RBD snapshot",
806                    bdrv_get_device_or_node_name(state->bs));
807         ret = -EINVAL;
808     }
810     return ret;
813 static void qemu_rbd_close(BlockDriverState *bs)
815     BDRVRBDState *s = bs->opaque;
817     rbd_close(s->image);
818     rados_ioctx_destroy(s->io_ctx);
819     g_free(s->snap);
820     g_free(s->image_name);
821     rados_shutdown(s->cluster);
824 /* Resize the RBD image and update the 'image_size' with the current size */
825 static int qemu_rbd_resize(BlockDriverState *bs, uint64_t size)
827     BDRVRBDState *s = bs->opaque;
828     int r;
830     r = rbd_resize(s->image, size);
831     if (r < 0) {
832         return r;
833     }
835     s->image_size = size;
837     return 0;
840 static const AIOCBInfo rbd_aiocb_info = {
841     .aiocb_size = sizeof(RBDAIOCB),
844 static void rbd_finish_bh(void *opaque)
846     RADOSCB *rcb = opaque;
847     qemu_rbd_complete_aio(rcb);
851  * This is the callback function for rbd_aio_read and _write
853  * Note: this function is being called from a non qemu thread so
854  * we need to be careful about what we do here. Generally we only
855  * schedule a BH, and do the rest of the io completion handling
856  * from rbd_finish_bh() which runs in a qemu context.
857  */
858 static void rbd_finish_aiocb(rbd_completion_t c, RADOSCB *rcb)
860     RBDAIOCB *acb = rcb->acb;
862     rcb->ret = rbd_aio_get_return_value(c);
863     rbd_aio_release(c);
865     replay_bh_schedule_oneshot_event(bdrv_get_aio_context(acb->common.bs),
866                                      rbd_finish_bh, rcb);
869 static int rbd_aio_discard_wrapper(rbd_image_t image,
870                                    uint64_t off,
871                                    uint64_t len,
872                                    rbd_completion_t comp)
874 #ifdef LIBRBD_SUPPORTS_DISCARD
875     return rbd_aio_discard(image, off, len, comp);
876 #else
877     return -ENOTSUP;
878 #endif
881 static int rbd_aio_flush_wrapper(rbd_image_t image,
882                                  rbd_completion_t comp)
884 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
885     return rbd_aio_flush(image, comp);
886 #else
887     return -ENOTSUP;
888 #endif
891 static BlockAIOCB *rbd_start_aio(BlockDriverState *bs,
892                                  int64_t off,
893                                  QEMUIOVector *qiov,
894                                  int64_t size,
895                                  BlockCompletionFunc *cb,
896                                  void *opaque,
897                                  RBDAIOCmd cmd)
899     RBDAIOCB *acb;
900     RADOSCB *rcb = NULL;
901     rbd_completion_t c;
902     int r;
904     BDRVRBDState *s = bs->opaque;
906     acb = qemu_aio_get(&rbd_aiocb_info, bs, cb, opaque);
907     acb->cmd = cmd;
908     acb->qiov = qiov;
909     assert(!qiov || qiov->size == size);
911     rcb = g_new(RADOSCB, 1);
913     if (!LIBRBD_USE_IOVEC) {
914         if (cmd == RBD_AIO_DISCARD || cmd == RBD_AIO_FLUSH) {
915             acb->bounce = NULL;
916         } else {
917             acb->bounce = qemu_try_blockalign(bs, qiov->size);
918             if (acb->bounce == NULL) {
919                 goto failed;
920             }
921         }
922         if (cmd == RBD_AIO_WRITE) {
923             qemu_iovec_to_buf(acb->qiov, 0, acb->bounce, qiov->size);
924         }
925         rcb->buf = acb->bounce;
926     }
928     acb->ret = 0;
929     acb->error = 0;
930     acb->s = s;
932     rcb->acb = acb;
933     rcb->s = acb->s;
934     rcb->size = size;
935     r = rbd_aio_create_completion(rcb, (rbd_callback_t) rbd_finish_aiocb, &c);
936     if (r < 0) {
937         goto failed;
938     }
940     switch (cmd) {
941     case RBD_AIO_WRITE: {
942         /*
943          * RBD APIs don't allow us to write more than actual size, so in order
944          * to support growing images, we resize the image before write
945          * operations that exceed the current size.
946          */
947         if (off + size > s->image_size) {
948             r = qemu_rbd_resize(bs, off + size);
949             if (r < 0) {
950                 goto failed_completion;
951             }
952         }
953 #ifdef LIBRBD_SUPPORTS_IOVEC
954             r = rbd_aio_writev(s->image, qiov->iov, qiov->niov, off, c);
955 #else
956             r = rbd_aio_write(s->image, off, size, rcb->buf, c);
957 #endif
958         break;
959     }
960     case RBD_AIO_READ:
961 #ifdef LIBRBD_SUPPORTS_IOVEC
962             r = rbd_aio_readv(s->image, qiov->iov, qiov->niov, off, c);
963 #else
964             r = rbd_aio_read(s->image, off, size, rcb->buf, c);
965 #endif
966         break;
967     case RBD_AIO_DISCARD:
968         r = rbd_aio_discard_wrapper(s->image, off, size, c);
969         break;
970     case RBD_AIO_FLUSH:
971         r = rbd_aio_flush_wrapper(s->image, c);
972         break;
973     default:
974         r = -EINVAL;
975     }
977     if (r < 0) {
978         goto failed_completion;
979     }
980     return &acb->common;
982 failed_completion:
983     rbd_aio_release(c);
984 failed:
985     g_free(rcb);
986     if (!LIBRBD_USE_IOVEC) {
987         qemu_vfree(acb->bounce);
988     }
990     qemu_aio_unref(acb);
991     return NULL;
994 static BlockAIOCB *qemu_rbd_aio_preadv(BlockDriverState *bs,
995                                        uint64_t offset, uint64_t bytes,
996                                        QEMUIOVector *qiov, int flags,
997                                        BlockCompletionFunc *cb,
998                                        void *opaque)
1000     return rbd_start_aio(bs, offset, qiov, bytes, cb, opaque,
1001                          RBD_AIO_READ);
1004 static BlockAIOCB *qemu_rbd_aio_pwritev(BlockDriverState *bs,
1005                                         uint64_t offset, uint64_t bytes,
1006                                         QEMUIOVector *qiov, int flags,
1007                                         BlockCompletionFunc *cb,
1008                                         void *opaque)
1010     return rbd_start_aio(bs, offset, qiov, bytes, cb, opaque,
1011                          RBD_AIO_WRITE);
1014 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
1015 static BlockAIOCB *qemu_rbd_aio_flush(BlockDriverState *bs,
1016                                       BlockCompletionFunc *cb,
1017                                       void *opaque)
1019     return rbd_start_aio(bs, 0, NULL, 0, cb, opaque, RBD_AIO_FLUSH);
1022 #else
1024 static int qemu_rbd_co_flush(BlockDriverState *bs)
1026 #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 1)
1027     /* rbd_flush added in 0.1.1 */
1028     BDRVRBDState *s = bs->opaque;
1029     return rbd_flush(s->image);
1030 #else
1031     return 0;
1032 #endif
1034 #endif
1036 static int qemu_rbd_getinfo(BlockDriverState *bs, BlockDriverInfo *bdi)
1038     BDRVRBDState *s = bs->opaque;
1039     rbd_image_info_t info;
1040     int r;
1042     r = rbd_stat(s->image, &info, sizeof(info));
1043     if (r < 0) {
1044         return r;
1045     }
1047     bdi->cluster_size = info.obj_size;
1048     return 0;
1051 static int64_t qemu_rbd_getlength(BlockDriverState *bs)
1053     BDRVRBDState *s = bs->opaque;
1054     rbd_image_info_t info;
1055     int r;
1057     r = rbd_stat(s->image, &info, sizeof(info));
1058     if (r < 0) {
1059         return r;
1060     }
1062     return info.size;
1065 static int coroutine_fn qemu_rbd_co_truncate(BlockDriverState *bs,
1066                                              int64_t offset,
1067                                              bool exact,
1068                                              PreallocMode prealloc,
1069                                              BdrvRequestFlags flags,
1070                                              Error **errp)
1072     int r;
1074     if (prealloc != PREALLOC_MODE_OFF) {
1075         error_setg(errp, "Unsupported preallocation mode '%s'",
1076                    PreallocMode_str(prealloc));
1077         return -ENOTSUP;
1078     }
1080     r = qemu_rbd_resize(bs, offset);
1081     if (r < 0) {
1082         error_setg_errno(errp, -r, "Failed to resize file");
1083         return r;
1084     }
1086     return 0;
1089 static int qemu_rbd_snap_create(BlockDriverState *bs,
1090                                 QEMUSnapshotInfo *sn_info)
1092     BDRVRBDState *s = bs->opaque;
1093     int r;
1095     if (sn_info->name[0] == '\0') {
1096         return -EINVAL; /* we need a name for rbd snapshots */
1097     }
1099     /*
1100      * rbd snapshots are using the name as the user controlled unique identifier
1101      * we can't use the rbd snapid for that purpose, as it can't be set
1102      */
1103     if (sn_info->id_str[0] != '\0' &&
1104         strcmp(sn_info->id_str, sn_info->name) != 0) {
1105         return -EINVAL;
1106     }
1108     if (strlen(sn_info->name) >= sizeof(sn_info->id_str)) {
1109         return -ERANGE;
1110     }
1112     r = rbd_snap_create(s->image, sn_info->name);
1113     if (r < 0) {
1114         error_report("failed to create snap: %s", strerror(-r));
1115         return r;
1116     }
1118     return 0;
1121 static int qemu_rbd_snap_remove(BlockDriverState *bs,
1122                                 const char *snapshot_id,
1123                                 const char *snapshot_name,
1124                                 Error **errp)
1126     BDRVRBDState *s = bs->opaque;
1127     int r;
1129     if (!snapshot_name) {
1130         error_setg(errp, "rbd need a valid snapshot name");
1131         return -EINVAL;
1132     }
1134     /* If snapshot_id is specified, it must be equal to name, see
1135        qemu_rbd_snap_list() */
1136     if (snapshot_id && strcmp(snapshot_id, snapshot_name)) {
1137         error_setg(errp,
1138                    "rbd do not support snapshot id, it should be NULL or "
1139                    "equal to snapshot name");
1140         return -EINVAL;
1141     }
1143     r = rbd_snap_remove(s->image, snapshot_name);
1144     if (r < 0) {
1145         error_setg_errno(errp, -r, "Failed to remove the snapshot");
1146     }
1147     return r;
1150 static int qemu_rbd_snap_rollback(BlockDriverState *bs,
1151                                   const char *snapshot_name)
1153     BDRVRBDState *s = bs->opaque;
1155     return rbd_snap_rollback(s->image, snapshot_name);
1158 static int qemu_rbd_snap_list(BlockDriverState *bs,
1159                               QEMUSnapshotInfo **psn_tab)
1161     BDRVRBDState *s = bs->opaque;
1162     QEMUSnapshotInfo *sn_info, *sn_tab = NULL;
1163     int i, snap_count;
1164     rbd_snap_info_t *snaps;
1165     int max_snaps = RBD_MAX_SNAPS;
1167     do {
1168         snaps = g_new(rbd_snap_info_t, max_snaps);
1169         snap_count = rbd_snap_list(s->image, snaps, &max_snaps);
1170         if (snap_count <= 0) {
1171             g_free(snaps);
1172         }
1173     } while (snap_count == -ERANGE);
1175     if (snap_count <= 0) {
1176         goto done;
1177     }
1179     sn_tab = g_new0(QEMUSnapshotInfo, snap_count);
1181     for (i = 0; i < snap_count; i++) {
1182         const char *snap_name = snaps[i].name;
1184         sn_info = sn_tab + i;
1185         pstrcpy(sn_info->id_str, sizeof(sn_info->id_str), snap_name);
1186         pstrcpy(sn_info->name, sizeof(sn_info->name), snap_name);
1188         sn_info->vm_state_size = snaps[i].size;
1189         sn_info->date_sec = 0;
1190         sn_info->date_nsec = 0;
1191         sn_info->vm_clock_nsec = 0;
1192     }
1193     rbd_snap_list_end(snaps);
1194     g_free(snaps);
1196  done:
1197     *psn_tab = sn_tab;
1198     return snap_count;
1201 #ifdef LIBRBD_SUPPORTS_DISCARD
1202 static BlockAIOCB *qemu_rbd_aio_pdiscard(BlockDriverState *bs,
1203                                          int64_t offset,
1204                                          int bytes,
1205                                          BlockCompletionFunc *cb,
1206                                          void *opaque)
1208     return rbd_start_aio(bs, offset, NULL, bytes, cb, opaque,
1209                          RBD_AIO_DISCARD);
1211 #endif
1213 #ifdef LIBRBD_SUPPORTS_INVALIDATE
1214 static void coroutine_fn qemu_rbd_co_invalidate_cache(BlockDriverState *bs,
1215                                                       Error **errp)
1217     BDRVRBDState *s = bs->opaque;
1218     int r = rbd_invalidate_cache(s->image);
1219     if (r < 0) {
1220         error_setg_errno(errp, -r, "Failed to invalidate the cache");
1221     }
1223 #endif
1225 static QemuOptsList qemu_rbd_create_opts = {
1226     .name = "rbd-create-opts",
1227     .head = QTAILQ_HEAD_INITIALIZER(qemu_rbd_create_opts.head),
1228     .desc = {
1229         {
1230             .name = BLOCK_OPT_SIZE,
1231             .type = QEMU_OPT_SIZE,
1232             .help = "Virtual disk size"
1233         },
1234         {
1235             .name = BLOCK_OPT_CLUSTER_SIZE,
1236             .type = QEMU_OPT_SIZE,
1237             .help = "RBD object size"
1238         },
1239         {
1240             .name = "password-secret",
1241             .type = QEMU_OPT_STRING,
1242             .help = "ID of secret providing the password",
1243         },
1244         { /* end of list */ }
1245     }
1248 static const char *const qemu_rbd_strong_runtime_opts[] = {
1249     "pool",
1250     "namespace",
1251     "image",
1252     "conf",
1253     "snapshot",
1254     "user",
1255     "server.",
1256     "password-secret",
1258     NULL
1261 static BlockDriver bdrv_rbd = {
1262     .format_name            = "rbd",
1263     .instance_size          = sizeof(BDRVRBDState),
1264     .bdrv_parse_filename    = qemu_rbd_parse_filename,
1265     .bdrv_refresh_limits    = qemu_rbd_refresh_limits,
1266     .bdrv_file_open         = qemu_rbd_open,
1267     .bdrv_close             = qemu_rbd_close,
1268     .bdrv_reopen_prepare    = qemu_rbd_reopen_prepare,
1269     .bdrv_co_create         = qemu_rbd_co_create,
1270     .bdrv_co_create_opts    = qemu_rbd_co_create_opts,
1271     .bdrv_has_zero_init     = bdrv_has_zero_init_1,
1272     .bdrv_get_info          = qemu_rbd_getinfo,
1273     .create_opts            = &qemu_rbd_create_opts,
1274     .bdrv_getlength         = qemu_rbd_getlength,
1275     .bdrv_co_truncate       = qemu_rbd_co_truncate,
1276     .protocol_name          = "rbd",
1278     .bdrv_aio_preadv        = qemu_rbd_aio_preadv,
1279     .bdrv_aio_pwritev       = qemu_rbd_aio_pwritev,
1281 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
1282     .bdrv_aio_flush         = qemu_rbd_aio_flush,
1283 #else
1284     .bdrv_co_flush_to_disk  = qemu_rbd_co_flush,
1285 #endif
1287 #ifdef LIBRBD_SUPPORTS_DISCARD
1288     .bdrv_aio_pdiscard      = qemu_rbd_aio_pdiscard,
1289 #endif
1291     .bdrv_snapshot_create   = qemu_rbd_snap_create,
1292     .bdrv_snapshot_delete   = qemu_rbd_snap_remove,
1293     .bdrv_snapshot_list     = qemu_rbd_snap_list,
1294     .bdrv_snapshot_goto     = qemu_rbd_snap_rollback,
1295 #ifdef LIBRBD_SUPPORTS_INVALIDATE
1296     .bdrv_co_invalidate_cache = qemu_rbd_co_invalidate_cache,
1297 #endif
1299     .strong_runtime_opts    = qemu_rbd_strong_runtime_opts,
1302 static void bdrv_rbd_init(void)
1304     bdrv_register(&bdrv_rbd);
1307 block_init(bdrv_rbd_init);