]> git.proxmox.com Git - mirror_qemu.git/blob - block/rbd.c
qapi: Replace qobject_to_X(o) by qobject_to(X, o)
[mirror_qemu.git] / block / rbd.c
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 */
13
14 #include "qemu/osdep.h"
15
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"
27 #include "qapi/qobject-input-visitor.h"
28 #include "qapi/qapi-visit-block-core.h"
29
30 /*
31 * When specifying the image filename use:
32 *
33 * rbd:poolname/devicename[@snapshotname][:option1=value1[:option2=value2...]]
34 *
35 * poolname must be the name of an existing rados pool.
36 *
37 * devicename is the name of the rbd image.
38 *
39 * Each option given is used to configure rados, and may be any valid
40 * Ceph option, "id", or "conf".
41 *
42 * The "id" option indicates what user we should authenticate as to
43 * the Ceph cluster. If it is excluded we will use the Ceph default
44 * (normally 'admin').
45 *
46 * The "conf" option specifies a Ceph configuration file to read. If
47 * it is not specified, we will read from the default Ceph locations
48 * (e.g., /etc/ceph/ceph.conf). To avoid reading _any_ configuration
49 * file, specify conf=/dev/null.
50 *
51 * Configuration values containing :, @, or = can be escaped with a
52 * leading "\".
53 */
54
55 /* rbd_aio_discard added in 0.1.2 */
56 #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 2)
57 #define LIBRBD_SUPPORTS_DISCARD
58 #else
59 #undef LIBRBD_SUPPORTS_DISCARD
60 #endif
61
62 #define OBJ_MAX_SIZE (1UL << OBJ_DEFAULT_OBJ_ORDER)
63
64 #define RBD_MAX_SNAPS 100
65
66 /* The LIBRBD_SUPPORTS_IOVEC is defined in librbd.h */
67 #ifdef LIBRBD_SUPPORTS_IOVEC
68 #define LIBRBD_USE_IOVEC 1
69 #else
70 #define LIBRBD_USE_IOVEC 0
71 #endif
72
73 typedef enum {
74 RBD_AIO_READ,
75 RBD_AIO_WRITE,
76 RBD_AIO_DISCARD,
77 RBD_AIO_FLUSH
78 } RBDAIOCmd;
79
80 typedef struct RBDAIOCB {
81 BlockAIOCB common;
82 int64_t ret;
83 QEMUIOVector *qiov;
84 char *bounce;
85 RBDAIOCmd cmd;
86 int error;
87 struct BDRVRBDState *s;
88 } RBDAIOCB;
89
90 typedef struct RADOSCB {
91 RBDAIOCB *acb;
92 struct BDRVRBDState *s;
93 int64_t size;
94 char *buf;
95 int64_t ret;
96 } RADOSCB;
97
98 typedef struct BDRVRBDState {
99 rados_t cluster;
100 rados_ioctx_t io_ctx;
101 rbd_image_t image;
102 char *image_name;
103 char *snap;
104 } BDRVRBDState;
105
106 static int qemu_rbd_connect(rados_t *cluster, rados_ioctx_t *io_ctx,
107 BlockdevOptionsRbd *opts, bool cache,
108 const char *keypairs, const char *secretid,
109 Error **errp);
110
111 static char *qemu_rbd_next_tok(char *src, char delim, char **p)
112 {
113 char *end;
114
115 *p = NULL;
116
117 for (end = src; *end; ++end) {
118 if (*end == delim) {
119 break;
120 }
121 if (*end == '\\' && end[1] != '\0') {
122 end++;
123 }
124 }
125 if (*end == delim) {
126 *p = end + 1;
127 *end = '\0';
128 }
129 return src;
130 }
131
132 static void qemu_rbd_unescape(char *src)
133 {
134 char *p;
135
136 for (p = src; *src; ++src, ++p) {
137 if (*src == '\\' && src[1] != '\0') {
138 src++;
139 }
140 *p = *src;
141 }
142 *p = '\0';
143 }
144
145 static void qemu_rbd_parse_filename(const char *filename, QDict *options,
146 Error **errp)
147 {
148 const char *start;
149 char *p, *buf;
150 QList *keypairs = NULL;
151 char *found_str;
152
153 if (!strstart(filename, "rbd:", &start)) {
154 error_setg(errp, "File name must start with 'rbd:'");
155 return;
156 }
157
158 buf = g_strdup(start);
159 p = buf;
160
161 found_str = qemu_rbd_next_tok(p, '/', &p);
162 if (!p) {
163 error_setg(errp, "Pool name is required");
164 goto done;
165 }
166 qemu_rbd_unescape(found_str);
167 qdict_put_str(options, "pool", found_str);
168
169 if (strchr(p, '@')) {
170 found_str = qemu_rbd_next_tok(p, '@', &p);
171 qemu_rbd_unescape(found_str);
172 qdict_put_str(options, "image", found_str);
173
174 found_str = qemu_rbd_next_tok(p, ':', &p);
175 qemu_rbd_unescape(found_str);
176 qdict_put_str(options, "snapshot", found_str);
177 } else {
178 found_str = qemu_rbd_next_tok(p, ':', &p);
179 qemu_rbd_unescape(found_str);
180 qdict_put_str(options, "image", found_str);
181 }
182 if (!p) {
183 goto done;
184 }
185
186 /* The following are essentially all key/value pairs, and we treat
187 * 'id' and 'conf' a bit special. Key/value pairs may be in any order. */
188 while (p) {
189 char *name, *value;
190 name = qemu_rbd_next_tok(p, '=', &p);
191 if (!p) {
192 error_setg(errp, "conf option %s has no value", name);
193 break;
194 }
195
196 qemu_rbd_unescape(name);
197
198 value = qemu_rbd_next_tok(p, ':', &p);
199 qemu_rbd_unescape(value);
200
201 if (!strcmp(name, "conf")) {
202 qdict_put_str(options, "conf", value);
203 } else if (!strcmp(name, "id")) {
204 qdict_put_str(options, "user", value);
205 } else {
206 /*
207 * We pass these internally to qemu_rbd_set_keypairs(), so
208 * we can get away with the simpler list of [ "key1",
209 * "value1", "key2", "value2" ] rather than a raw dict
210 * { "key1": "value1", "key2": "value2" } where we can't
211 * guarantee order, or even a more correct but complex
212 * [ { "key1": "value1" }, { "key2": "value2" } ]
213 */
214 if (!keypairs) {
215 keypairs = qlist_new();
216 }
217 qlist_append_str(keypairs, name);
218 qlist_append_str(keypairs, value);
219 }
220 }
221
222 if (keypairs) {
223 qdict_put(options, "=keyvalue-pairs",
224 qobject_to_json(QOBJECT(keypairs)));
225 }
226
227 done:
228 g_free(buf);
229 QDECREF(keypairs);
230 return;
231 }
232
233
234 static int qemu_rbd_set_auth(rados_t cluster, const char *secretid,
235 Error **errp)
236 {
237 if (secretid == 0) {
238 return 0;
239 }
240
241 gchar *secret = qcrypto_secret_lookup_as_base64(secretid,
242 errp);
243 if (!secret) {
244 return -1;
245 }
246
247 rados_conf_set(cluster, "key", secret);
248 g_free(secret);
249
250 return 0;
251 }
252
253 static int qemu_rbd_set_keypairs(rados_t cluster, const char *keypairs_json,
254 Error **errp)
255 {
256 QList *keypairs;
257 QString *name;
258 QString *value;
259 const char *key;
260 size_t remaining;
261 int ret = 0;
262
263 if (!keypairs_json) {
264 return ret;
265 }
266 keypairs = qobject_to(QList,
267 qobject_from_json(keypairs_json, &error_abort));
268 remaining = qlist_size(keypairs) / 2;
269 assert(remaining);
270
271 while (remaining--) {
272 name = qobject_to(QString, qlist_pop(keypairs));
273 value = qobject_to(QString, qlist_pop(keypairs));
274 assert(name && value);
275 key = qstring_get_str(name);
276
277 ret = rados_conf_set(cluster, key, qstring_get_str(value));
278 QDECREF(value);
279 if (ret < 0) {
280 error_setg_errno(errp, -ret, "invalid conf option %s", key);
281 QDECREF(name);
282 ret = -EINVAL;
283 break;
284 }
285 QDECREF(name);
286 }
287
288 QDECREF(keypairs);
289 return ret;
290 }
291
292 static void qemu_rbd_memset(RADOSCB *rcb, int64_t offs)
293 {
294 if (LIBRBD_USE_IOVEC) {
295 RBDAIOCB *acb = rcb->acb;
296 iov_memset(acb->qiov->iov, acb->qiov->niov, offs, 0,
297 acb->qiov->size - offs);
298 } else {
299 memset(rcb->buf + offs, 0, rcb->size - offs);
300 }
301 }
302
303 static QemuOptsList runtime_opts = {
304 .name = "rbd",
305 .head = QTAILQ_HEAD_INITIALIZER(runtime_opts.head),
306 .desc = {
307 {
308 .name = "pool",
309 .type = QEMU_OPT_STRING,
310 .help = "Rados pool name",
311 },
312 {
313 .name = "image",
314 .type = QEMU_OPT_STRING,
315 .help = "Image name in the pool",
316 },
317 {
318 .name = "conf",
319 .type = QEMU_OPT_STRING,
320 .help = "Rados config file location",
321 },
322 {
323 .name = "snapshot",
324 .type = QEMU_OPT_STRING,
325 .help = "Ceph snapshot name",
326 },
327 {
328 /* maps to 'id' in rados_create() */
329 .name = "user",
330 .type = QEMU_OPT_STRING,
331 .help = "Rados id name",
332 },
333 /*
334 * server.* extracted manually, see qemu_rbd_mon_host()
335 */
336 { /* end of list */ }
337 },
338 };
339
340 /* FIXME Deprecate and remove keypairs or make it available in QMP.
341 * password_secret should eventually be configurable in opts->location. Support
342 * for it in .bdrv_open will make it work here as well. */
343 static int qemu_rbd_do_create(BlockdevCreateOptions *options,
344 const char *keypairs, const char *password_secret,
345 Error **errp)
346 {
347 BlockdevCreateOptionsRbd *opts = &options->u.rbd;
348 rados_t cluster;
349 rados_ioctx_t io_ctx;
350 int obj_order = 0;
351 int ret;
352
353 assert(options->driver == BLOCKDEV_DRIVER_RBD);
354 if (opts->location->has_snapshot) {
355 error_setg(errp, "Can't use snapshot name for image creation");
356 return -EINVAL;
357 }
358
359 if (opts->has_cluster_size) {
360 int64_t objsize = opts->cluster_size;
361 if ((objsize - 1) & objsize) { /* not a power of 2? */
362 error_setg(errp, "obj size needs to be power of 2");
363 return -EINVAL;
364 }
365 if (objsize < 4096) {
366 error_setg(errp, "obj size too small");
367 return -EINVAL;
368 }
369 obj_order = ctz32(objsize);
370 }
371
372 ret = qemu_rbd_connect(&cluster, &io_ctx, opts->location, false, keypairs,
373 password_secret, errp);
374 if (ret < 0) {
375 return ret;
376 }
377
378 ret = rbd_create(io_ctx, opts->location->image, opts->size, &obj_order);
379 if (ret < 0) {
380 error_setg_errno(errp, -ret, "error rbd create");
381 goto out;
382 }
383
384 ret = 0;
385 out:
386 rados_ioctx_destroy(io_ctx);
387 rados_shutdown(cluster);
388 return ret;
389 }
390
391 static int qemu_rbd_co_create(BlockdevCreateOptions *options, Error **errp)
392 {
393 return qemu_rbd_do_create(options, NULL, NULL, errp);
394 }
395
396 static int coroutine_fn qemu_rbd_co_create_opts(const char *filename,
397 QemuOpts *opts,
398 Error **errp)
399 {
400 BlockdevCreateOptions *create_options;
401 BlockdevCreateOptionsRbd *rbd_opts;
402 BlockdevOptionsRbd *loc;
403 Error *local_err = NULL;
404 const char *keypairs, *password_secret;
405 QDict *options = NULL;
406 int ret = 0;
407
408 create_options = g_new0(BlockdevCreateOptions, 1);
409 create_options->driver = BLOCKDEV_DRIVER_RBD;
410 rbd_opts = &create_options->u.rbd;
411
412 rbd_opts->location = g_new0(BlockdevOptionsRbd, 1);
413
414 password_secret = qemu_opt_get(opts, "password-secret");
415
416 /* Read out options */
417 rbd_opts->size = ROUND_UP(qemu_opt_get_size_del(opts, BLOCK_OPT_SIZE, 0),
418 BDRV_SECTOR_SIZE);
419 rbd_opts->cluster_size = qemu_opt_get_size_del(opts,
420 BLOCK_OPT_CLUSTER_SIZE, 0);
421 rbd_opts->has_cluster_size = (rbd_opts->cluster_size != 0);
422
423 options = qdict_new();
424 qemu_rbd_parse_filename(filename, options, &local_err);
425 if (local_err) {
426 ret = -EINVAL;
427 error_propagate(errp, local_err);
428 goto exit;
429 }
430
431 /*
432 * Caution: while qdict_get_try_str() is fine, getting non-string
433 * types would require more care. When @options come from -blockdev
434 * or blockdev_add, its members are typed according to the QAPI
435 * schema, but when they come from -drive, they're all QString.
436 */
437 loc = rbd_opts->location;
438 loc->pool = g_strdup(qdict_get_try_str(options, "pool"));
439 loc->conf = g_strdup(qdict_get_try_str(options, "conf"));
440 loc->has_conf = !!loc->conf;
441 loc->user = g_strdup(qdict_get_try_str(options, "user"));
442 loc->has_user = !!loc->user;
443 loc->image = g_strdup(qdict_get_try_str(options, "image"));
444 keypairs = qdict_get_try_str(options, "=keyvalue-pairs");
445
446 ret = qemu_rbd_do_create(create_options, keypairs, password_secret, errp);
447 if (ret < 0) {
448 goto exit;
449 }
450
451 exit:
452 QDECREF(options);
453 qapi_free_BlockdevCreateOptions(create_options);
454 return ret;
455 }
456
457 /*
458 * This aio completion is being called from rbd_finish_bh() and runs in qemu
459 * BH context.
460 */
461 static void qemu_rbd_complete_aio(RADOSCB *rcb)
462 {
463 RBDAIOCB *acb = rcb->acb;
464 int64_t r;
465
466 r = rcb->ret;
467
468 if (acb->cmd != RBD_AIO_READ) {
469 if (r < 0) {
470 acb->ret = r;
471 acb->error = 1;
472 } else if (!acb->error) {
473 acb->ret = rcb->size;
474 }
475 } else {
476 if (r < 0) {
477 qemu_rbd_memset(rcb, 0);
478 acb->ret = r;
479 acb->error = 1;
480 } else if (r < rcb->size) {
481 qemu_rbd_memset(rcb, r);
482 if (!acb->error) {
483 acb->ret = rcb->size;
484 }
485 } else if (!acb->error) {
486 acb->ret = r;
487 }
488 }
489
490 g_free(rcb);
491
492 if (!LIBRBD_USE_IOVEC) {
493 if (acb->cmd == RBD_AIO_READ) {
494 qemu_iovec_from_buf(acb->qiov, 0, acb->bounce, acb->qiov->size);
495 }
496 qemu_vfree(acb->bounce);
497 }
498
499 acb->common.cb(acb->common.opaque, (acb->ret > 0 ? 0 : acb->ret));
500
501 qemu_aio_unref(acb);
502 }
503
504 static char *qemu_rbd_mon_host(BlockdevOptionsRbd *opts, Error **errp)
505 {
506 const char **vals;
507 const char *host, *port;
508 char *rados_str;
509 InetSocketAddressBaseList *p;
510 int i, cnt;
511
512 if (!opts->has_server) {
513 return NULL;
514 }
515
516 for (cnt = 0, p = opts->server; p; p = p->next) {
517 cnt++;
518 }
519
520 vals = g_new(const char *, cnt + 1);
521
522 for (i = 0, p = opts->server; p; p = p->next, i++) {
523 host = p->value->host;
524 port = p->value->port;
525
526 if (strchr(host, ':')) {
527 vals[i] = g_strdup_printf("[%s]:%s", host, port);
528 } else {
529 vals[i] = g_strdup_printf("%s:%s", host, port);
530 }
531 }
532 vals[i] = NULL;
533
534 rados_str = i ? g_strjoinv(";", (char **)vals) : NULL;
535 g_strfreev((char **)vals);
536 return rados_str;
537 }
538
539 static int qemu_rbd_connect(rados_t *cluster, rados_ioctx_t *io_ctx,
540 BlockdevOptionsRbd *opts, bool cache,
541 const char *keypairs, const char *secretid,
542 Error **errp)
543 {
544 char *mon_host = NULL;
545 Error *local_err = NULL;
546 int r;
547
548 mon_host = qemu_rbd_mon_host(opts, &local_err);
549 if (local_err) {
550 error_propagate(errp, local_err);
551 r = -EINVAL;
552 goto failed_opts;
553 }
554
555 r = rados_create(cluster, opts->user);
556 if (r < 0) {
557 error_setg_errno(errp, -r, "error initializing");
558 goto failed_opts;
559 }
560
561 /* try default location when conf=NULL, but ignore failure */
562 r = rados_conf_read_file(*cluster, opts->conf);
563 if (opts->has_conf && r < 0) {
564 error_setg_errno(errp, -r, "error reading conf file %s", opts->conf);
565 goto failed_shutdown;
566 }
567
568 r = qemu_rbd_set_keypairs(*cluster, keypairs, errp);
569 if (r < 0) {
570 goto failed_shutdown;
571 }
572
573 if (mon_host) {
574 r = rados_conf_set(*cluster, "mon_host", mon_host);
575 if (r < 0) {
576 goto failed_shutdown;
577 }
578 }
579
580 if (qemu_rbd_set_auth(*cluster, secretid, errp) < 0) {
581 r = -EIO;
582 goto failed_shutdown;
583 }
584
585 /*
586 * Fallback to more conservative semantics if setting cache
587 * options fails. Ignore errors from setting rbd_cache because the
588 * only possible error is that the option does not exist, and
589 * librbd defaults to no caching. If write through caching cannot
590 * be set up, fall back to no caching.
591 */
592 if (cache) {
593 rados_conf_set(*cluster, "rbd_cache", "true");
594 } else {
595 rados_conf_set(*cluster, "rbd_cache", "false");
596 }
597
598 r = rados_connect(*cluster);
599 if (r < 0) {
600 error_setg_errno(errp, -r, "error connecting");
601 goto failed_shutdown;
602 }
603
604 r = rados_ioctx_create(*cluster, opts->pool, io_ctx);
605 if (r < 0) {
606 error_setg_errno(errp, -r, "error opening pool %s", opts->pool);
607 goto failed_shutdown;
608 }
609
610 return 0;
611
612 failed_shutdown:
613 rados_shutdown(*cluster);
614 failed_opts:
615 g_free(mon_host);
616 return r;
617 }
618
619 static int qemu_rbd_open(BlockDriverState *bs, QDict *options, int flags,
620 Error **errp)
621 {
622 BDRVRBDState *s = bs->opaque;
623 BlockdevOptionsRbd *opts = NULL;
624 Visitor *v;
625 QObject *crumpled = NULL;
626 Error *local_err = NULL;
627 const char *filename;
628 char *keypairs, *secretid;
629 int r;
630
631 /* If we are given a filename, parse the filename, with precedence given to
632 * filename encoded options */
633 filename = qdict_get_try_str(options, "filename");
634 if (filename) {
635 warn_report("'filename' option specified. "
636 "This is an unsupported option, and may be deprecated "
637 "in the future");
638 qemu_rbd_parse_filename(filename, options, &local_err);
639 qdict_del(options, "filename");
640 if (local_err) {
641 error_propagate(errp, local_err);
642 return -EINVAL;
643 }
644 }
645
646 keypairs = g_strdup(qdict_get_try_str(options, "=keyvalue-pairs"));
647 if (keypairs) {
648 qdict_del(options, "=keyvalue-pairs");
649 }
650
651 secretid = g_strdup(qdict_get_try_str(options, "password-secret"));
652 if (secretid) {
653 qdict_del(options, "password-secret");
654 }
655
656 /* Convert the remaining options into a QAPI object */
657 crumpled = qdict_crumple(options, errp);
658 if (crumpled == NULL) {
659 r = -EINVAL;
660 goto out;
661 }
662
663 v = qobject_input_visitor_new_keyval(crumpled);
664 visit_type_BlockdevOptionsRbd(v, NULL, &opts, &local_err);
665 visit_free(v);
666 qobject_decref(crumpled);
667
668 if (local_err) {
669 error_propagate(errp, local_err);
670 r = -EINVAL;
671 goto out;
672 }
673
674 r = qemu_rbd_connect(&s->cluster, &s->io_ctx, opts,
675 !(flags & BDRV_O_NOCACHE), keypairs, secretid, errp);
676 if (r < 0) {
677 goto out;
678 }
679
680 s->snap = g_strdup(opts->snapshot);
681 s->image_name = g_strdup(opts->image);
682
683 /* rbd_open is always r/w */
684 r = rbd_open(s->io_ctx, s->image_name, &s->image, s->snap);
685 if (r < 0) {
686 error_setg_errno(errp, -r, "error reading header from %s",
687 s->image_name);
688 goto failed_open;
689 }
690
691 /* If we are using an rbd snapshot, we must be r/o, otherwise
692 * leave as-is */
693 if (s->snap != NULL) {
694 if (!bdrv_is_read_only(bs)) {
695 error_report("Opening rbd snapshots without an explicit "
696 "read-only=on option is deprecated. Future versions "
697 "will refuse to open the image instead of "
698 "automatically marking the image read-only.");
699 r = bdrv_set_read_only(bs, true, &local_err);
700 if (r < 0) {
701 error_propagate(errp, local_err);
702 goto failed_open;
703 }
704 }
705 }
706
707 r = 0;
708 goto out;
709
710 failed_open:
711 rados_ioctx_destroy(s->io_ctx);
712 g_free(s->snap);
713 g_free(s->image_name);
714 rados_shutdown(s->cluster);
715 out:
716 qapi_free_BlockdevOptionsRbd(opts);
717 g_free(keypairs);
718 g_free(secretid);
719 return r;
720 }
721
722
723 /* Since RBD is currently always opened R/W via the API,
724 * we just need to check if we are using a snapshot or not, in
725 * order to determine if we will allow it to be R/W */
726 static int qemu_rbd_reopen_prepare(BDRVReopenState *state,
727 BlockReopenQueue *queue, Error **errp)
728 {
729 BDRVRBDState *s = state->bs->opaque;
730 int ret = 0;
731
732 if (s->snap && state->flags & BDRV_O_RDWR) {
733 error_setg(errp,
734 "Cannot change node '%s' to r/w when using RBD snapshot",
735 bdrv_get_device_or_node_name(state->bs));
736 ret = -EINVAL;
737 }
738
739 return ret;
740 }
741
742 static void qemu_rbd_close(BlockDriverState *bs)
743 {
744 BDRVRBDState *s = bs->opaque;
745
746 rbd_close(s->image);
747 rados_ioctx_destroy(s->io_ctx);
748 g_free(s->snap);
749 g_free(s->image_name);
750 rados_shutdown(s->cluster);
751 }
752
753 static const AIOCBInfo rbd_aiocb_info = {
754 .aiocb_size = sizeof(RBDAIOCB),
755 };
756
757 static void rbd_finish_bh(void *opaque)
758 {
759 RADOSCB *rcb = opaque;
760 qemu_rbd_complete_aio(rcb);
761 }
762
763 /*
764 * This is the callback function for rbd_aio_read and _write
765 *
766 * Note: this function is being called from a non qemu thread so
767 * we need to be careful about what we do here. Generally we only
768 * schedule a BH, and do the rest of the io completion handling
769 * from rbd_finish_bh() which runs in a qemu context.
770 */
771 static void rbd_finish_aiocb(rbd_completion_t c, RADOSCB *rcb)
772 {
773 RBDAIOCB *acb = rcb->acb;
774
775 rcb->ret = rbd_aio_get_return_value(c);
776 rbd_aio_release(c);
777
778 aio_bh_schedule_oneshot(bdrv_get_aio_context(acb->common.bs),
779 rbd_finish_bh, rcb);
780 }
781
782 static int rbd_aio_discard_wrapper(rbd_image_t image,
783 uint64_t off,
784 uint64_t len,
785 rbd_completion_t comp)
786 {
787 #ifdef LIBRBD_SUPPORTS_DISCARD
788 return rbd_aio_discard(image, off, len, comp);
789 #else
790 return -ENOTSUP;
791 #endif
792 }
793
794 static int rbd_aio_flush_wrapper(rbd_image_t image,
795 rbd_completion_t comp)
796 {
797 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
798 return rbd_aio_flush(image, comp);
799 #else
800 return -ENOTSUP;
801 #endif
802 }
803
804 static BlockAIOCB *rbd_start_aio(BlockDriverState *bs,
805 int64_t off,
806 QEMUIOVector *qiov,
807 int64_t size,
808 BlockCompletionFunc *cb,
809 void *opaque,
810 RBDAIOCmd cmd)
811 {
812 RBDAIOCB *acb;
813 RADOSCB *rcb = NULL;
814 rbd_completion_t c;
815 int r;
816
817 BDRVRBDState *s = bs->opaque;
818
819 acb = qemu_aio_get(&rbd_aiocb_info, bs, cb, opaque);
820 acb->cmd = cmd;
821 acb->qiov = qiov;
822 assert(!qiov || qiov->size == size);
823
824 rcb = g_new(RADOSCB, 1);
825
826 if (!LIBRBD_USE_IOVEC) {
827 if (cmd == RBD_AIO_DISCARD || cmd == RBD_AIO_FLUSH) {
828 acb->bounce = NULL;
829 } else {
830 acb->bounce = qemu_try_blockalign(bs, qiov->size);
831 if (acb->bounce == NULL) {
832 goto failed;
833 }
834 }
835 if (cmd == RBD_AIO_WRITE) {
836 qemu_iovec_to_buf(acb->qiov, 0, acb->bounce, qiov->size);
837 }
838 rcb->buf = acb->bounce;
839 }
840
841 acb->ret = 0;
842 acb->error = 0;
843 acb->s = s;
844
845 rcb->acb = acb;
846 rcb->s = acb->s;
847 rcb->size = size;
848 r = rbd_aio_create_completion(rcb, (rbd_callback_t) rbd_finish_aiocb, &c);
849 if (r < 0) {
850 goto failed;
851 }
852
853 switch (cmd) {
854 case RBD_AIO_WRITE:
855 #ifdef LIBRBD_SUPPORTS_IOVEC
856 r = rbd_aio_writev(s->image, qiov->iov, qiov->niov, off, c);
857 #else
858 r = rbd_aio_write(s->image, off, size, rcb->buf, c);
859 #endif
860 break;
861 case RBD_AIO_READ:
862 #ifdef LIBRBD_SUPPORTS_IOVEC
863 r = rbd_aio_readv(s->image, qiov->iov, qiov->niov, off, c);
864 #else
865 r = rbd_aio_read(s->image, off, size, rcb->buf, c);
866 #endif
867 break;
868 case RBD_AIO_DISCARD:
869 r = rbd_aio_discard_wrapper(s->image, off, size, c);
870 break;
871 case RBD_AIO_FLUSH:
872 r = rbd_aio_flush_wrapper(s->image, c);
873 break;
874 default:
875 r = -EINVAL;
876 }
877
878 if (r < 0) {
879 goto failed_completion;
880 }
881 return &acb->common;
882
883 failed_completion:
884 rbd_aio_release(c);
885 failed:
886 g_free(rcb);
887 if (!LIBRBD_USE_IOVEC) {
888 qemu_vfree(acb->bounce);
889 }
890
891 qemu_aio_unref(acb);
892 return NULL;
893 }
894
895 static BlockAIOCB *qemu_rbd_aio_readv(BlockDriverState *bs,
896 int64_t sector_num,
897 QEMUIOVector *qiov,
898 int nb_sectors,
899 BlockCompletionFunc *cb,
900 void *opaque)
901 {
902 return rbd_start_aio(bs, sector_num << BDRV_SECTOR_BITS, qiov,
903 (int64_t) nb_sectors << BDRV_SECTOR_BITS, cb, opaque,
904 RBD_AIO_READ);
905 }
906
907 static BlockAIOCB *qemu_rbd_aio_writev(BlockDriverState *bs,
908 int64_t sector_num,
909 QEMUIOVector *qiov,
910 int nb_sectors,
911 BlockCompletionFunc *cb,
912 void *opaque)
913 {
914 return rbd_start_aio(bs, sector_num << BDRV_SECTOR_BITS, qiov,
915 (int64_t) nb_sectors << BDRV_SECTOR_BITS, cb, opaque,
916 RBD_AIO_WRITE);
917 }
918
919 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
920 static BlockAIOCB *qemu_rbd_aio_flush(BlockDriverState *bs,
921 BlockCompletionFunc *cb,
922 void *opaque)
923 {
924 return rbd_start_aio(bs, 0, NULL, 0, cb, opaque, RBD_AIO_FLUSH);
925 }
926
927 #else
928
929 static int qemu_rbd_co_flush(BlockDriverState *bs)
930 {
931 #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 1)
932 /* rbd_flush added in 0.1.1 */
933 BDRVRBDState *s = bs->opaque;
934 return rbd_flush(s->image);
935 #else
936 return 0;
937 #endif
938 }
939 #endif
940
941 static int qemu_rbd_getinfo(BlockDriverState *bs, BlockDriverInfo *bdi)
942 {
943 BDRVRBDState *s = bs->opaque;
944 rbd_image_info_t info;
945 int r;
946
947 r = rbd_stat(s->image, &info, sizeof(info));
948 if (r < 0) {
949 return r;
950 }
951
952 bdi->cluster_size = info.obj_size;
953 return 0;
954 }
955
956 static int64_t qemu_rbd_getlength(BlockDriverState *bs)
957 {
958 BDRVRBDState *s = bs->opaque;
959 rbd_image_info_t info;
960 int r;
961
962 r = rbd_stat(s->image, &info, sizeof(info));
963 if (r < 0) {
964 return r;
965 }
966
967 return info.size;
968 }
969
970 static int qemu_rbd_truncate(BlockDriverState *bs, int64_t offset,
971 PreallocMode prealloc, Error **errp)
972 {
973 BDRVRBDState *s = bs->opaque;
974 int r;
975
976 if (prealloc != PREALLOC_MODE_OFF) {
977 error_setg(errp, "Unsupported preallocation mode '%s'",
978 PreallocMode_str(prealloc));
979 return -ENOTSUP;
980 }
981
982 r = rbd_resize(s->image, offset);
983 if (r < 0) {
984 error_setg_errno(errp, -r, "Failed to resize file");
985 return r;
986 }
987
988 return 0;
989 }
990
991 static int qemu_rbd_snap_create(BlockDriverState *bs,
992 QEMUSnapshotInfo *sn_info)
993 {
994 BDRVRBDState *s = bs->opaque;
995 int r;
996
997 if (sn_info->name[0] == '\0') {
998 return -EINVAL; /* we need a name for rbd snapshots */
999 }
1000
1001 /*
1002 * rbd snapshots are using the name as the user controlled unique identifier
1003 * we can't use the rbd snapid for that purpose, as it can't be set
1004 */
1005 if (sn_info->id_str[0] != '\0' &&
1006 strcmp(sn_info->id_str, sn_info->name) != 0) {
1007 return -EINVAL;
1008 }
1009
1010 if (strlen(sn_info->name) >= sizeof(sn_info->id_str)) {
1011 return -ERANGE;
1012 }
1013
1014 r = rbd_snap_create(s->image, sn_info->name);
1015 if (r < 0) {
1016 error_report("failed to create snap: %s", strerror(-r));
1017 return r;
1018 }
1019
1020 return 0;
1021 }
1022
1023 static int qemu_rbd_snap_remove(BlockDriverState *bs,
1024 const char *snapshot_id,
1025 const char *snapshot_name,
1026 Error **errp)
1027 {
1028 BDRVRBDState *s = bs->opaque;
1029 int r;
1030
1031 if (!snapshot_name) {
1032 error_setg(errp, "rbd need a valid snapshot name");
1033 return -EINVAL;
1034 }
1035
1036 /* If snapshot_id is specified, it must be equal to name, see
1037 qemu_rbd_snap_list() */
1038 if (snapshot_id && strcmp(snapshot_id, snapshot_name)) {
1039 error_setg(errp,
1040 "rbd do not support snapshot id, it should be NULL or "
1041 "equal to snapshot name");
1042 return -EINVAL;
1043 }
1044
1045 r = rbd_snap_remove(s->image, snapshot_name);
1046 if (r < 0) {
1047 error_setg_errno(errp, -r, "Failed to remove the snapshot");
1048 }
1049 return r;
1050 }
1051
1052 static int qemu_rbd_snap_rollback(BlockDriverState *bs,
1053 const char *snapshot_name)
1054 {
1055 BDRVRBDState *s = bs->opaque;
1056
1057 return rbd_snap_rollback(s->image, snapshot_name);
1058 }
1059
1060 static int qemu_rbd_snap_list(BlockDriverState *bs,
1061 QEMUSnapshotInfo **psn_tab)
1062 {
1063 BDRVRBDState *s = bs->opaque;
1064 QEMUSnapshotInfo *sn_info, *sn_tab = NULL;
1065 int i, snap_count;
1066 rbd_snap_info_t *snaps;
1067 int max_snaps = RBD_MAX_SNAPS;
1068
1069 do {
1070 snaps = g_new(rbd_snap_info_t, max_snaps);
1071 snap_count = rbd_snap_list(s->image, snaps, &max_snaps);
1072 if (snap_count <= 0) {
1073 g_free(snaps);
1074 }
1075 } while (snap_count == -ERANGE);
1076
1077 if (snap_count <= 0) {
1078 goto done;
1079 }
1080
1081 sn_tab = g_new0(QEMUSnapshotInfo, snap_count);
1082
1083 for (i = 0; i < snap_count; i++) {
1084 const char *snap_name = snaps[i].name;
1085
1086 sn_info = sn_tab + i;
1087 pstrcpy(sn_info->id_str, sizeof(sn_info->id_str), snap_name);
1088 pstrcpy(sn_info->name, sizeof(sn_info->name), snap_name);
1089
1090 sn_info->vm_state_size = snaps[i].size;
1091 sn_info->date_sec = 0;
1092 sn_info->date_nsec = 0;
1093 sn_info->vm_clock_nsec = 0;
1094 }
1095 rbd_snap_list_end(snaps);
1096 g_free(snaps);
1097
1098 done:
1099 *psn_tab = sn_tab;
1100 return snap_count;
1101 }
1102
1103 #ifdef LIBRBD_SUPPORTS_DISCARD
1104 static BlockAIOCB *qemu_rbd_aio_pdiscard(BlockDriverState *bs,
1105 int64_t offset,
1106 int bytes,
1107 BlockCompletionFunc *cb,
1108 void *opaque)
1109 {
1110 return rbd_start_aio(bs, offset, NULL, bytes, cb, opaque,
1111 RBD_AIO_DISCARD);
1112 }
1113 #endif
1114
1115 #ifdef LIBRBD_SUPPORTS_INVALIDATE
1116 static void coroutine_fn qemu_rbd_co_invalidate_cache(BlockDriverState *bs,
1117 Error **errp)
1118 {
1119 BDRVRBDState *s = bs->opaque;
1120 int r = rbd_invalidate_cache(s->image);
1121 if (r < 0) {
1122 error_setg_errno(errp, -r, "Failed to invalidate the cache");
1123 }
1124 }
1125 #endif
1126
1127 static QemuOptsList qemu_rbd_create_opts = {
1128 .name = "rbd-create-opts",
1129 .head = QTAILQ_HEAD_INITIALIZER(qemu_rbd_create_opts.head),
1130 .desc = {
1131 {
1132 .name = BLOCK_OPT_SIZE,
1133 .type = QEMU_OPT_SIZE,
1134 .help = "Virtual disk size"
1135 },
1136 {
1137 .name = BLOCK_OPT_CLUSTER_SIZE,
1138 .type = QEMU_OPT_SIZE,
1139 .help = "RBD object size"
1140 },
1141 {
1142 .name = "password-secret",
1143 .type = QEMU_OPT_STRING,
1144 .help = "ID of secret providing the password",
1145 },
1146 { /* end of list */ }
1147 }
1148 };
1149
1150 static BlockDriver bdrv_rbd = {
1151 .format_name = "rbd",
1152 .instance_size = sizeof(BDRVRBDState),
1153 .bdrv_parse_filename = qemu_rbd_parse_filename,
1154 .bdrv_file_open = qemu_rbd_open,
1155 .bdrv_close = qemu_rbd_close,
1156 .bdrv_reopen_prepare = qemu_rbd_reopen_prepare,
1157 .bdrv_co_create = qemu_rbd_co_create,
1158 .bdrv_co_create_opts = qemu_rbd_co_create_opts,
1159 .bdrv_has_zero_init = bdrv_has_zero_init_1,
1160 .bdrv_get_info = qemu_rbd_getinfo,
1161 .create_opts = &qemu_rbd_create_opts,
1162 .bdrv_getlength = qemu_rbd_getlength,
1163 .bdrv_truncate = qemu_rbd_truncate,
1164 .protocol_name = "rbd",
1165
1166 .bdrv_aio_readv = qemu_rbd_aio_readv,
1167 .bdrv_aio_writev = qemu_rbd_aio_writev,
1168
1169 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
1170 .bdrv_aio_flush = qemu_rbd_aio_flush,
1171 #else
1172 .bdrv_co_flush_to_disk = qemu_rbd_co_flush,
1173 #endif
1174
1175 #ifdef LIBRBD_SUPPORTS_DISCARD
1176 .bdrv_aio_pdiscard = qemu_rbd_aio_pdiscard,
1177 #endif
1178
1179 .bdrv_snapshot_create = qemu_rbd_snap_create,
1180 .bdrv_snapshot_delete = qemu_rbd_snap_remove,
1181 .bdrv_snapshot_list = qemu_rbd_snap_list,
1182 .bdrv_snapshot_goto = qemu_rbd_snap_rollback,
1183 #ifdef LIBRBD_SUPPORTS_INVALIDATE
1184 .bdrv_co_invalidate_cache = qemu_rbd_co_invalidate_cache,
1185 #endif
1186 };
1187
1188 static void bdrv_rbd_init(void)
1189 {
1190 bdrv_register(&bdrv_rbd);
1191 }
1192
1193 block_init(bdrv_rbd_init);