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