Merge remote-tracking branch 'remotes/kevin/tags/for-upstream' into staging
[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 <inttypes.h>
15
16 #include "qemu-common.h"
17 #include "qemu/error-report.h"
18 #include "block/block_int.h"
19
20 #include <rbd/librbd.h>
21
22 /*
23 * When specifying the image filename use:
24 *
25 * rbd:poolname/devicename[@snapshotname][:option1=value1[:option2=value2...]]
26 *
27 * poolname must be the name of an existing rados pool.
28 *
29 * devicename is the name of the rbd image.
30 *
31 * Each option given is used to configure rados, and may be any valid
32 * Ceph option, "id", or "conf".
33 *
34 * The "id" option indicates what user we should authenticate as to
35 * the Ceph cluster. If it is excluded we will use the Ceph default
36 * (normally 'admin').
37 *
38 * The "conf" option specifies a Ceph configuration file to read. If
39 * it is not specified, we will read from the default Ceph locations
40 * (e.g., /etc/ceph/ceph.conf). To avoid reading _any_ configuration
41 * file, specify conf=/dev/null.
42 *
43 * Configuration values containing :, @, or = can be escaped with a
44 * leading "\".
45 */
46
47 /* rbd_aio_discard added in 0.1.2 */
48 #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 2)
49 #define LIBRBD_SUPPORTS_DISCARD
50 #else
51 #undef LIBRBD_SUPPORTS_DISCARD
52 #endif
53
54 #define OBJ_MAX_SIZE (1UL << OBJ_DEFAULT_OBJ_ORDER)
55
56 #define RBD_MAX_CONF_NAME_SIZE 128
57 #define RBD_MAX_CONF_VAL_SIZE 512
58 #define RBD_MAX_CONF_SIZE 1024
59 #define RBD_MAX_POOL_NAME_SIZE 128
60 #define RBD_MAX_SNAP_NAME_SIZE 128
61 #define RBD_MAX_SNAPS 100
62
63 typedef enum {
64 RBD_AIO_READ,
65 RBD_AIO_WRITE,
66 RBD_AIO_DISCARD,
67 RBD_AIO_FLUSH
68 } RBDAIOCmd;
69
70 typedef struct RBDAIOCB {
71 BlockDriverAIOCB common;
72 QEMUBH *bh;
73 int64_t ret;
74 QEMUIOVector *qiov;
75 char *bounce;
76 RBDAIOCmd cmd;
77 int64_t sector_num;
78 int error;
79 struct BDRVRBDState *s;
80 int cancelled;
81 int status;
82 } RBDAIOCB;
83
84 typedef struct RADOSCB {
85 int rcbid;
86 RBDAIOCB *acb;
87 struct BDRVRBDState *s;
88 int done;
89 int64_t size;
90 char *buf;
91 int64_t ret;
92 } RADOSCB;
93
94 #define RBD_FD_READ 0
95 #define RBD_FD_WRITE 1
96
97 typedef struct BDRVRBDState {
98 rados_t cluster;
99 rados_ioctx_t io_ctx;
100 rbd_image_t image;
101 char name[RBD_MAX_IMAGE_NAME_SIZE];
102 char *snap;
103 } BDRVRBDState;
104
105 static int qemu_rbd_next_tok(char *dst, int dst_len,
106 char *src, char delim,
107 const char *name,
108 char **p)
109 {
110 int l;
111 char *end;
112
113 *p = NULL;
114
115 if (delim != '\0') {
116 for (end = src; *end; ++end) {
117 if (*end == delim) {
118 break;
119 }
120 if (*end == '\\' && end[1] != '\0') {
121 end++;
122 }
123 }
124 if (*end == delim) {
125 *p = end + 1;
126 *end = '\0';
127 }
128 }
129 l = strlen(src);
130 if (l >= dst_len) {
131 error_report("%s too long", name);
132 return -EINVAL;
133 } else if (l == 0) {
134 error_report("%s too short", name);
135 return -EINVAL;
136 }
137
138 pstrcpy(dst, dst_len, src);
139
140 return 0;
141 }
142
143 static void qemu_rbd_unescape(char *src)
144 {
145 char *p;
146
147 for (p = src; *src; ++src, ++p) {
148 if (*src == '\\' && src[1] != '\0') {
149 src++;
150 }
151 *p = *src;
152 }
153 *p = '\0';
154 }
155
156 static int qemu_rbd_parsename(const char *filename,
157 char *pool, int pool_len,
158 char *snap, int snap_len,
159 char *name, int name_len,
160 char *conf, int conf_len)
161 {
162 const char *start;
163 char *p, *buf;
164 int ret;
165
166 if (!strstart(filename, "rbd:", &start)) {
167 return -EINVAL;
168 }
169
170 buf = g_strdup(start);
171 p = buf;
172 *snap = '\0';
173 *conf = '\0';
174
175 ret = qemu_rbd_next_tok(pool, pool_len, p, '/', "pool name", &p);
176 if (ret < 0 || !p) {
177 ret = -EINVAL;
178 goto done;
179 }
180 qemu_rbd_unescape(pool);
181
182 if (strchr(p, '@')) {
183 ret = qemu_rbd_next_tok(name, name_len, p, '@', "object name", &p);
184 if (ret < 0) {
185 goto done;
186 }
187 ret = qemu_rbd_next_tok(snap, snap_len, p, ':', "snap name", &p);
188 qemu_rbd_unescape(snap);
189 } else {
190 ret = qemu_rbd_next_tok(name, name_len, p, ':', "object name", &p);
191 }
192 qemu_rbd_unescape(name);
193 if (ret < 0 || !p) {
194 goto done;
195 }
196
197 ret = qemu_rbd_next_tok(conf, conf_len, p, '\0', "configuration", &p);
198
199 done:
200 g_free(buf);
201 return ret;
202 }
203
204 static char *qemu_rbd_parse_clientname(const char *conf, char *clientname)
205 {
206 const char *p = conf;
207
208 while (*p) {
209 int len;
210 const char *end = strchr(p, ':');
211
212 if (end) {
213 len = end - p;
214 } else {
215 len = strlen(p);
216 }
217
218 if (strncmp(p, "id=", 3) == 0) {
219 len -= 3;
220 strncpy(clientname, p + 3, len);
221 clientname[len] = '\0';
222 return clientname;
223 }
224 if (end == NULL) {
225 break;
226 }
227 p = end + 1;
228 }
229 return NULL;
230 }
231
232 static int qemu_rbd_set_conf(rados_t cluster, const char *conf)
233 {
234 char *p, *buf;
235 char name[RBD_MAX_CONF_NAME_SIZE];
236 char value[RBD_MAX_CONF_VAL_SIZE];
237 int ret = 0;
238
239 buf = g_strdup(conf);
240 p = buf;
241
242 while (p) {
243 ret = qemu_rbd_next_tok(name, sizeof(name), p,
244 '=', "conf option name", &p);
245 if (ret < 0) {
246 break;
247 }
248 qemu_rbd_unescape(name);
249
250 if (!p) {
251 error_report("conf option %s has no value", name);
252 ret = -EINVAL;
253 break;
254 }
255
256 ret = qemu_rbd_next_tok(value, sizeof(value), p,
257 ':', "conf option value", &p);
258 if (ret < 0) {
259 break;
260 }
261 qemu_rbd_unescape(value);
262
263 if (strcmp(name, "conf") == 0) {
264 ret = rados_conf_read_file(cluster, value);
265 if (ret < 0) {
266 error_report("error reading conf file %s", value);
267 break;
268 }
269 } else if (strcmp(name, "id") == 0) {
270 /* ignore, this is parsed by qemu_rbd_parse_clientname() */
271 } else {
272 ret = rados_conf_set(cluster, name, value);
273 if (ret < 0) {
274 error_report("invalid conf option %s", name);
275 ret = -EINVAL;
276 break;
277 }
278 }
279 }
280
281 g_free(buf);
282 return ret;
283 }
284
285 static int qemu_rbd_create(const char *filename, QEMUOptionParameter *options,
286 Error **errp)
287 {
288 int64_t bytes = 0;
289 int64_t objsize;
290 int obj_order = 0;
291 char pool[RBD_MAX_POOL_NAME_SIZE];
292 char name[RBD_MAX_IMAGE_NAME_SIZE];
293 char snap_buf[RBD_MAX_SNAP_NAME_SIZE];
294 char conf[RBD_MAX_CONF_SIZE];
295 char clientname_buf[RBD_MAX_CONF_SIZE];
296 char *clientname;
297 rados_t cluster;
298 rados_ioctx_t io_ctx;
299 int ret;
300
301 if (qemu_rbd_parsename(filename, pool, sizeof(pool),
302 snap_buf, sizeof(snap_buf),
303 name, sizeof(name),
304 conf, sizeof(conf)) < 0) {
305 return -EINVAL;
306 }
307
308 /* Read out options */
309 while (options && options->name) {
310 if (!strcmp(options->name, BLOCK_OPT_SIZE)) {
311 bytes = options->value.n;
312 } else if (!strcmp(options->name, BLOCK_OPT_CLUSTER_SIZE)) {
313 if (options->value.n) {
314 objsize = options->value.n;
315 if ((objsize - 1) & objsize) { /* not a power of 2? */
316 error_report("obj size needs to be power of 2");
317 return -EINVAL;
318 }
319 if (objsize < 4096) {
320 error_report("obj size too small");
321 return -EINVAL;
322 }
323 obj_order = ffs(objsize) - 1;
324 }
325 }
326 options++;
327 }
328
329 clientname = qemu_rbd_parse_clientname(conf, clientname_buf);
330 if (rados_create(&cluster, clientname) < 0) {
331 error_report("error initializing");
332 return -EIO;
333 }
334
335 if (strstr(conf, "conf=") == NULL) {
336 /* try default location, but ignore failure */
337 rados_conf_read_file(cluster, NULL);
338 }
339
340 if (conf[0] != '\0' &&
341 qemu_rbd_set_conf(cluster, conf) < 0) {
342 error_report("error setting config options");
343 rados_shutdown(cluster);
344 return -EIO;
345 }
346
347 if (rados_connect(cluster) < 0) {
348 error_report("error connecting");
349 rados_shutdown(cluster);
350 return -EIO;
351 }
352
353 if (rados_ioctx_create(cluster, pool, &io_ctx) < 0) {
354 error_report("error opening pool %s", pool);
355 rados_shutdown(cluster);
356 return -EIO;
357 }
358
359 ret = rbd_create(io_ctx, name, bytes, &obj_order);
360 rados_ioctx_destroy(io_ctx);
361 rados_shutdown(cluster);
362
363 return ret;
364 }
365
366 /*
367 * This aio completion is being called from rbd_finish_bh() and runs in qemu
368 * BH context.
369 */
370 static void qemu_rbd_complete_aio(RADOSCB *rcb)
371 {
372 RBDAIOCB *acb = rcb->acb;
373 int64_t r;
374
375 r = rcb->ret;
376
377 if (acb->cmd != RBD_AIO_READ) {
378 if (r < 0) {
379 acb->ret = r;
380 acb->error = 1;
381 } else if (!acb->error) {
382 acb->ret = rcb->size;
383 }
384 } else {
385 if (r < 0) {
386 memset(rcb->buf, 0, rcb->size);
387 acb->ret = r;
388 acb->error = 1;
389 } else if (r < rcb->size) {
390 memset(rcb->buf + r, 0, rcb->size - r);
391 if (!acb->error) {
392 acb->ret = rcb->size;
393 }
394 } else if (!acb->error) {
395 acb->ret = r;
396 }
397 }
398
399 g_free(rcb);
400
401 if (acb->cmd == RBD_AIO_READ) {
402 qemu_iovec_from_buf(acb->qiov, 0, acb->bounce, acb->qiov->size);
403 }
404 qemu_vfree(acb->bounce);
405 acb->common.cb(acb->common.opaque, (acb->ret > 0 ? 0 : acb->ret));
406 acb->status = 0;
407
408 if (!acb->cancelled) {
409 qemu_aio_release(acb);
410 }
411 }
412
413 /* TODO Convert to fine grained options */
414 static QemuOptsList runtime_opts = {
415 .name = "rbd",
416 .head = QTAILQ_HEAD_INITIALIZER(runtime_opts.head),
417 .desc = {
418 {
419 .name = "filename",
420 .type = QEMU_OPT_STRING,
421 .help = "Specification of the rbd image",
422 },
423 { /* end of list */ }
424 },
425 };
426
427 static int qemu_rbd_open(BlockDriverState *bs, QDict *options, int flags,
428 Error **errp)
429 {
430 BDRVRBDState *s = bs->opaque;
431 char pool[RBD_MAX_POOL_NAME_SIZE];
432 char snap_buf[RBD_MAX_SNAP_NAME_SIZE];
433 char conf[RBD_MAX_CONF_SIZE];
434 char clientname_buf[RBD_MAX_CONF_SIZE];
435 char *clientname;
436 QemuOpts *opts;
437 Error *local_err = NULL;
438 const char *filename;
439 int r;
440
441 opts = qemu_opts_create(&runtime_opts, NULL, 0, &error_abort);
442 qemu_opts_absorb_qdict(opts, options, &local_err);
443 if (local_err) {
444 qerror_report_err(local_err);
445 error_free(local_err);
446 qemu_opts_del(opts);
447 return -EINVAL;
448 }
449
450 filename = qemu_opt_get(opts, "filename");
451
452 if (qemu_rbd_parsename(filename, pool, sizeof(pool),
453 snap_buf, sizeof(snap_buf),
454 s->name, sizeof(s->name),
455 conf, sizeof(conf)) < 0) {
456 r = -EINVAL;
457 goto failed_opts;
458 }
459
460 clientname = qemu_rbd_parse_clientname(conf, clientname_buf);
461 r = rados_create(&s->cluster, clientname);
462 if (r < 0) {
463 error_report("error initializing");
464 goto failed_opts;
465 }
466
467 s->snap = NULL;
468 if (snap_buf[0] != '\0') {
469 s->snap = g_strdup(snap_buf);
470 }
471
472 /*
473 * Fallback to more conservative semantics if setting cache
474 * options fails. Ignore errors from setting rbd_cache because the
475 * only possible error is that the option does not exist, and
476 * librbd defaults to no caching. If write through caching cannot
477 * be set up, fall back to no caching.
478 */
479 if (flags & BDRV_O_NOCACHE) {
480 rados_conf_set(s->cluster, "rbd_cache", "false");
481 } else {
482 rados_conf_set(s->cluster, "rbd_cache", "true");
483 }
484
485 if (strstr(conf, "conf=") == NULL) {
486 /* try default location, but ignore failure */
487 rados_conf_read_file(s->cluster, NULL);
488 }
489
490 if (conf[0] != '\0') {
491 r = qemu_rbd_set_conf(s->cluster, conf);
492 if (r < 0) {
493 error_report("error setting config options");
494 goto failed_shutdown;
495 }
496 }
497
498 r = rados_connect(s->cluster);
499 if (r < 0) {
500 error_report("error connecting");
501 goto failed_shutdown;
502 }
503
504 r = rados_ioctx_create(s->cluster, pool, &s->io_ctx);
505 if (r < 0) {
506 error_report("error opening pool %s", pool);
507 goto failed_shutdown;
508 }
509
510 r = rbd_open(s->io_ctx, s->name, &s->image, s->snap);
511 if (r < 0) {
512 error_report("error reading header from %s", s->name);
513 goto failed_open;
514 }
515
516 bs->read_only = (s->snap != NULL);
517
518 qemu_opts_del(opts);
519 return 0;
520
521 failed_open:
522 rados_ioctx_destroy(s->io_ctx);
523 failed_shutdown:
524 rados_shutdown(s->cluster);
525 g_free(s->snap);
526 failed_opts:
527 qemu_opts_del(opts);
528 return r;
529 }
530
531 static void qemu_rbd_close(BlockDriverState *bs)
532 {
533 BDRVRBDState *s = bs->opaque;
534
535 rbd_close(s->image);
536 rados_ioctx_destroy(s->io_ctx);
537 g_free(s->snap);
538 rados_shutdown(s->cluster);
539 }
540
541 /*
542 * Cancel aio. Since we don't reference acb in a non qemu threads,
543 * it is safe to access it here.
544 */
545 static void qemu_rbd_aio_cancel(BlockDriverAIOCB *blockacb)
546 {
547 RBDAIOCB *acb = (RBDAIOCB *) blockacb;
548 acb->cancelled = 1;
549
550 while (acb->status == -EINPROGRESS) {
551 qemu_aio_wait();
552 }
553
554 qemu_aio_release(acb);
555 }
556
557 static const AIOCBInfo rbd_aiocb_info = {
558 .aiocb_size = sizeof(RBDAIOCB),
559 .cancel = qemu_rbd_aio_cancel,
560 };
561
562 static void rbd_finish_bh(void *opaque)
563 {
564 RADOSCB *rcb = opaque;
565 qemu_bh_delete(rcb->acb->bh);
566 qemu_rbd_complete_aio(rcb);
567 }
568
569 /*
570 * This is the callback function for rbd_aio_read and _write
571 *
572 * Note: this function is being called from a non qemu thread so
573 * we need to be careful about what we do here. Generally we only
574 * schedule a BH, and do the rest of the io completion handling
575 * from rbd_finish_bh() which runs in a qemu context.
576 */
577 static void rbd_finish_aiocb(rbd_completion_t c, RADOSCB *rcb)
578 {
579 RBDAIOCB *acb = rcb->acb;
580
581 rcb->ret = rbd_aio_get_return_value(c);
582 rbd_aio_release(c);
583
584 acb->bh = qemu_bh_new(rbd_finish_bh, rcb);
585 qemu_bh_schedule(acb->bh);
586 }
587
588 static int rbd_aio_discard_wrapper(rbd_image_t image,
589 uint64_t off,
590 uint64_t len,
591 rbd_completion_t comp)
592 {
593 #ifdef LIBRBD_SUPPORTS_DISCARD
594 return rbd_aio_discard(image, off, len, comp);
595 #else
596 return -ENOTSUP;
597 #endif
598 }
599
600 static int rbd_aio_flush_wrapper(rbd_image_t image,
601 rbd_completion_t comp)
602 {
603 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
604 return rbd_aio_flush(image, comp);
605 #else
606 return -ENOTSUP;
607 #endif
608 }
609
610 static BlockDriverAIOCB *rbd_start_aio(BlockDriverState *bs,
611 int64_t sector_num,
612 QEMUIOVector *qiov,
613 int nb_sectors,
614 BlockDriverCompletionFunc *cb,
615 void *opaque,
616 RBDAIOCmd cmd)
617 {
618 RBDAIOCB *acb;
619 RADOSCB *rcb;
620 rbd_completion_t c;
621 int64_t off, size;
622 char *buf;
623 int r;
624
625 BDRVRBDState *s = bs->opaque;
626
627 acb = qemu_aio_get(&rbd_aiocb_info, bs, cb, opaque);
628 acb->cmd = cmd;
629 acb->qiov = qiov;
630 if (cmd == RBD_AIO_DISCARD || cmd == RBD_AIO_FLUSH) {
631 acb->bounce = NULL;
632 } else {
633 acb->bounce = qemu_blockalign(bs, qiov->size);
634 }
635 acb->ret = 0;
636 acb->error = 0;
637 acb->s = s;
638 acb->cancelled = 0;
639 acb->bh = NULL;
640 acb->status = -EINPROGRESS;
641
642 if (cmd == RBD_AIO_WRITE) {
643 qemu_iovec_to_buf(acb->qiov, 0, acb->bounce, qiov->size);
644 }
645
646 buf = acb->bounce;
647
648 off = sector_num * BDRV_SECTOR_SIZE;
649 size = nb_sectors * BDRV_SECTOR_SIZE;
650
651 rcb = g_malloc(sizeof(RADOSCB));
652 rcb->done = 0;
653 rcb->acb = acb;
654 rcb->buf = buf;
655 rcb->s = acb->s;
656 rcb->size = size;
657 r = rbd_aio_create_completion(rcb, (rbd_callback_t) rbd_finish_aiocb, &c);
658 if (r < 0) {
659 goto failed;
660 }
661
662 switch (cmd) {
663 case RBD_AIO_WRITE:
664 r = rbd_aio_write(s->image, off, size, buf, c);
665 break;
666 case RBD_AIO_READ:
667 r = rbd_aio_read(s->image, off, size, buf, c);
668 break;
669 case RBD_AIO_DISCARD:
670 r = rbd_aio_discard_wrapper(s->image, off, size, c);
671 break;
672 case RBD_AIO_FLUSH:
673 r = rbd_aio_flush_wrapper(s->image, c);
674 break;
675 default:
676 r = -EINVAL;
677 }
678
679 if (r < 0) {
680 goto failed;
681 }
682
683 return &acb->common;
684
685 failed:
686 g_free(rcb);
687 qemu_aio_release(acb);
688 return NULL;
689 }
690
691 static BlockDriverAIOCB *qemu_rbd_aio_readv(BlockDriverState *bs,
692 int64_t sector_num,
693 QEMUIOVector *qiov,
694 int nb_sectors,
695 BlockDriverCompletionFunc *cb,
696 void *opaque)
697 {
698 return rbd_start_aio(bs, sector_num, qiov, nb_sectors, cb, opaque,
699 RBD_AIO_READ);
700 }
701
702 static BlockDriverAIOCB *qemu_rbd_aio_writev(BlockDriverState *bs,
703 int64_t sector_num,
704 QEMUIOVector *qiov,
705 int nb_sectors,
706 BlockDriverCompletionFunc *cb,
707 void *opaque)
708 {
709 return rbd_start_aio(bs, sector_num, qiov, nb_sectors, cb, opaque,
710 RBD_AIO_WRITE);
711 }
712
713 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
714 static BlockDriverAIOCB *qemu_rbd_aio_flush(BlockDriverState *bs,
715 BlockDriverCompletionFunc *cb,
716 void *opaque)
717 {
718 return rbd_start_aio(bs, 0, NULL, 0, cb, opaque, RBD_AIO_FLUSH);
719 }
720
721 #else
722
723 static int qemu_rbd_co_flush(BlockDriverState *bs)
724 {
725 #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 1)
726 /* rbd_flush added in 0.1.1 */
727 BDRVRBDState *s = bs->opaque;
728 return rbd_flush(s->image);
729 #else
730 return 0;
731 #endif
732 }
733 #endif
734
735 static int qemu_rbd_getinfo(BlockDriverState *bs, BlockDriverInfo *bdi)
736 {
737 BDRVRBDState *s = bs->opaque;
738 rbd_image_info_t info;
739 int r;
740
741 r = rbd_stat(s->image, &info, sizeof(info));
742 if (r < 0) {
743 return r;
744 }
745
746 bdi->cluster_size = info.obj_size;
747 return 0;
748 }
749
750 static int64_t qemu_rbd_getlength(BlockDriverState *bs)
751 {
752 BDRVRBDState *s = bs->opaque;
753 rbd_image_info_t info;
754 int r;
755
756 r = rbd_stat(s->image, &info, sizeof(info));
757 if (r < 0) {
758 return r;
759 }
760
761 return info.size;
762 }
763
764 static int qemu_rbd_truncate(BlockDriverState *bs, int64_t offset)
765 {
766 BDRVRBDState *s = bs->opaque;
767 int r;
768
769 r = rbd_resize(s->image, offset);
770 if (r < 0) {
771 return r;
772 }
773
774 return 0;
775 }
776
777 static int qemu_rbd_snap_create(BlockDriverState *bs,
778 QEMUSnapshotInfo *sn_info)
779 {
780 BDRVRBDState *s = bs->opaque;
781 int r;
782
783 if (sn_info->name[0] == '\0') {
784 return -EINVAL; /* we need a name for rbd snapshots */
785 }
786
787 /*
788 * rbd snapshots are using the name as the user controlled unique identifier
789 * we can't use the rbd snapid for that purpose, as it can't be set
790 */
791 if (sn_info->id_str[0] != '\0' &&
792 strcmp(sn_info->id_str, sn_info->name) != 0) {
793 return -EINVAL;
794 }
795
796 if (strlen(sn_info->name) >= sizeof(sn_info->id_str)) {
797 return -ERANGE;
798 }
799
800 r = rbd_snap_create(s->image, sn_info->name);
801 if (r < 0) {
802 error_report("failed to create snap: %s", strerror(-r));
803 return r;
804 }
805
806 return 0;
807 }
808
809 static int qemu_rbd_snap_remove(BlockDriverState *bs,
810 const char *snapshot_id,
811 const char *snapshot_name,
812 Error **errp)
813 {
814 BDRVRBDState *s = bs->opaque;
815 int r;
816
817 if (!snapshot_name) {
818 error_setg(errp, "rbd need a valid snapshot name");
819 return -EINVAL;
820 }
821
822 /* If snapshot_id is specified, it must be equal to name, see
823 qemu_rbd_snap_list() */
824 if (snapshot_id && strcmp(snapshot_id, snapshot_name)) {
825 error_setg(errp,
826 "rbd do not support snapshot id, it should be NULL or "
827 "equal to snapshot name");
828 return -EINVAL;
829 }
830
831 r = rbd_snap_remove(s->image, snapshot_name);
832 if (r < 0) {
833 error_setg_errno(errp, -r, "Failed to remove the snapshot");
834 }
835 return r;
836 }
837
838 static int qemu_rbd_snap_rollback(BlockDriverState *bs,
839 const char *snapshot_name)
840 {
841 BDRVRBDState *s = bs->opaque;
842 int r;
843
844 r = rbd_snap_rollback(s->image, snapshot_name);
845 return r;
846 }
847
848 static int qemu_rbd_snap_list(BlockDriverState *bs,
849 QEMUSnapshotInfo **psn_tab)
850 {
851 BDRVRBDState *s = bs->opaque;
852 QEMUSnapshotInfo *sn_info, *sn_tab = NULL;
853 int i, snap_count;
854 rbd_snap_info_t *snaps;
855 int max_snaps = RBD_MAX_SNAPS;
856
857 do {
858 snaps = g_malloc(sizeof(*snaps) * max_snaps);
859 snap_count = rbd_snap_list(s->image, snaps, &max_snaps);
860 if (snap_count <= 0) {
861 g_free(snaps);
862 }
863 } while (snap_count == -ERANGE);
864
865 if (snap_count <= 0) {
866 goto done;
867 }
868
869 sn_tab = g_malloc0(snap_count * sizeof(QEMUSnapshotInfo));
870
871 for (i = 0; i < snap_count; i++) {
872 const char *snap_name = snaps[i].name;
873
874 sn_info = sn_tab + i;
875 pstrcpy(sn_info->id_str, sizeof(sn_info->id_str), snap_name);
876 pstrcpy(sn_info->name, sizeof(sn_info->name), snap_name);
877
878 sn_info->vm_state_size = snaps[i].size;
879 sn_info->date_sec = 0;
880 sn_info->date_nsec = 0;
881 sn_info->vm_clock_nsec = 0;
882 }
883 rbd_snap_list_end(snaps);
884 g_free(snaps);
885
886 done:
887 *psn_tab = sn_tab;
888 return snap_count;
889 }
890
891 #ifdef LIBRBD_SUPPORTS_DISCARD
892 static BlockDriverAIOCB* qemu_rbd_aio_discard(BlockDriverState *bs,
893 int64_t sector_num,
894 int nb_sectors,
895 BlockDriverCompletionFunc *cb,
896 void *opaque)
897 {
898 return rbd_start_aio(bs, sector_num, NULL, nb_sectors, cb, opaque,
899 RBD_AIO_DISCARD);
900 }
901 #endif
902
903 static QEMUOptionParameter qemu_rbd_create_options[] = {
904 {
905 .name = BLOCK_OPT_SIZE,
906 .type = OPT_SIZE,
907 .help = "Virtual disk size"
908 },
909 {
910 .name = BLOCK_OPT_CLUSTER_SIZE,
911 .type = OPT_SIZE,
912 .help = "RBD object size"
913 },
914 {NULL}
915 };
916
917 static BlockDriver bdrv_rbd = {
918 .format_name = "rbd",
919 .instance_size = sizeof(BDRVRBDState),
920 .bdrv_needs_filename = true,
921 .bdrv_file_open = qemu_rbd_open,
922 .bdrv_close = qemu_rbd_close,
923 .bdrv_create = qemu_rbd_create,
924 .bdrv_has_zero_init = bdrv_has_zero_init_1,
925 .bdrv_get_info = qemu_rbd_getinfo,
926 .create_options = qemu_rbd_create_options,
927 .bdrv_getlength = qemu_rbd_getlength,
928 .bdrv_truncate = qemu_rbd_truncate,
929 .protocol_name = "rbd",
930
931 .bdrv_aio_readv = qemu_rbd_aio_readv,
932 .bdrv_aio_writev = qemu_rbd_aio_writev,
933
934 #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
935 .bdrv_aio_flush = qemu_rbd_aio_flush,
936 #else
937 .bdrv_co_flush_to_disk = qemu_rbd_co_flush,
938 #endif
939
940 #ifdef LIBRBD_SUPPORTS_DISCARD
941 .bdrv_aio_discard = qemu_rbd_aio_discard,
942 #endif
943
944 .bdrv_snapshot_create = qemu_rbd_snap_create,
945 .bdrv_snapshot_delete = qemu_rbd_snap_remove,
946 .bdrv_snapshot_list = qemu_rbd_snap_list,
947 .bdrv_snapshot_goto = qemu_rbd_snap_rollback,
948 };
949
950 static void bdrv_rbd_init(void)
951 {
952 bdrv_register(&bdrv_rbd);
953 }
954
955 block_init(bdrv_rbd_init);