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