PPC: e500mc: add missing IVORs to bitmap
[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.h"
18 #include "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 #define OBJ_MAX_SIZE (1UL << OBJ_DEFAULT_OBJ_ORDER)
48
49 #define RBD_MAX_CONF_NAME_SIZE 128
50 #define RBD_MAX_CONF_VAL_SIZE 512
51 #define RBD_MAX_CONF_SIZE 1024
52 #define RBD_MAX_POOL_NAME_SIZE 128
53 #define RBD_MAX_SNAP_NAME_SIZE 128
54 #define RBD_MAX_SNAPS 100
55
56 typedef struct RBDAIOCB {
57 BlockDriverAIOCB common;
58 QEMUBH *bh;
59 int ret;
60 QEMUIOVector *qiov;
61 char *bounce;
62 int write;
63 int64_t sector_num;
64 int error;
65 struct BDRVRBDState *s;
66 int cancelled;
67 } RBDAIOCB;
68
69 typedef struct RADOSCB {
70 int rcbid;
71 RBDAIOCB *acb;
72 struct BDRVRBDState *s;
73 int done;
74 int64_t size;
75 char *buf;
76 int ret;
77 } RADOSCB;
78
79 #define RBD_FD_READ 0
80 #define RBD_FD_WRITE 1
81
82 typedef struct BDRVRBDState {
83 int fds[2];
84 rados_t cluster;
85 rados_ioctx_t io_ctx;
86 rbd_image_t image;
87 char name[RBD_MAX_IMAGE_NAME_SIZE];
88 int qemu_aio_count;
89 char *snap;
90 int event_reader_pos;
91 RADOSCB *event_rcb;
92 } BDRVRBDState;
93
94 static void rbd_aio_bh_cb(void *opaque);
95
96 static int qemu_rbd_next_tok(char *dst, int dst_len,
97 char *src, char delim,
98 const char *name,
99 char **p)
100 {
101 int l;
102 char *end;
103
104 *p = NULL;
105
106 if (delim != '\0') {
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 }
120 l = strlen(src);
121 if (l >= dst_len) {
122 error_report("%s too long", name);
123 return -EINVAL;
124 } else if (l == 0) {
125 error_report("%s too short", name);
126 return -EINVAL;
127 }
128
129 pstrcpy(dst, dst_len, src);
130
131 return 0;
132 }
133
134 static void qemu_rbd_unescape(char *src)
135 {
136 char *p;
137
138 for (p = src; *src; ++src, ++p) {
139 if (*src == '\\' && src[1] != '\0') {
140 src++;
141 }
142 *p = *src;
143 }
144 *p = '\0';
145 }
146
147 static int qemu_rbd_parsename(const char *filename,
148 char *pool, int pool_len,
149 char *snap, int snap_len,
150 char *name, int name_len,
151 char *conf, int conf_len)
152 {
153 const char *start;
154 char *p, *buf;
155 int ret;
156
157 if (!strstart(filename, "rbd:", &start)) {
158 return -EINVAL;
159 }
160
161 buf = g_strdup(start);
162 p = buf;
163 *snap = '\0';
164 *conf = '\0';
165
166 ret = qemu_rbd_next_tok(pool, pool_len, p, '/', "pool name", &p);
167 if (ret < 0 || !p) {
168 ret = -EINVAL;
169 goto done;
170 }
171 qemu_rbd_unescape(pool);
172
173 if (strchr(p, '@')) {
174 ret = qemu_rbd_next_tok(name, name_len, p, '@', "object name", &p);
175 if (ret < 0) {
176 goto done;
177 }
178 ret = qemu_rbd_next_tok(snap, snap_len, p, ':', "snap name", &p);
179 qemu_rbd_unescape(snap);
180 } else {
181 ret = qemu_rbd_next_tok(name, name_len, p, ':', "object name", &p);
182 }
183 qemu_rbd_unescape(name);
184 if (ret < 0 || !p) {
185 goto done;
186 }
187
188 ret = qemu_rbd_next_tok(conf, conf_len, p, '\0', "configuration", &p);
189
190 done:
191 g_free(buf);
192 return ret;
193 }
194
195 static char *qemu_rbd_parse_clientname(const char *conf, char *clientname)
196 {
197 const char *p = conf;
198
199 while (*p) {
200 int len;
201 const char *end = strchr(p, ':');
202
203 if (end) {
204 len = end - p;
205 } else {
206 len = strlen(p);
207 }
208
209 if (strncmp(p, "id=", 3) == 0) {
210 len -= 3;
211 strncpy(clientname, p + 3, len);
212 clientname[len] = '\0';
213 return clientname;
214 }
215 if (end == NULL) {
216 break;
217 }
218 p = end + 1;
219 }
220 return NULL;
221 }
222
223 static int qemu_rbd_set_conf(rados_t cluster, const char *conf)
224 {
225 char *p, *buf;
226 char name[RBD_MAX_CONF_NAME_SIZE];
227 char value[RBD_MAX_CONF_VAL_SIZE];
228 int ret = 0;
229
230 buf = g_strdup(conf);
231 p = buf;
232
233 while (p) {
234 ret = qemu_rbd_next_tok(name, sizeof(name), p,
235 '=', "conf option name", &p);
236 if (ret < 0) {
237 break;
238 }
239 qemu_rbd_unescape(name);
240
241 if (!p) {
242 error_report("conf option %s has no value", name);
243 ret = -EINVAL;
244 break;
245 }
246
247 ret = qemu_rbd_next_tok(value, sizeof(value), p,
248 ':', "conf option value", &p);
249 if (ret < 0) {
250 break;
251 }
252 qemu_rbd_unescape(value);
253
254 if (strcmp(name, "conf") == 0) {
255 ret = rados_conf_read_file(cluster, value);
256 if (ret < 0) {
257 error_report("error reading conf file %s", value);
258 break;
259 }
260 } else if (strcmp(name, "id") == 0) {
261 /* ignore, this is parsed by qemu_rbd_parse_clientname() */
262 } else {
263 ret = rados_conf_set(cluster, name, value);
264 if (ret < 0) {
265 error_report("invalid conf option %s", name);
266 ret = -EINVAL;
267 break;
268 }
269 }
270 }
271
272 g_free(buf);
273 return ret;
274 }
275
276 static int qemu_rbd_create(const char *filename, QEMUOptionParameter *options)
277 {
278 int64_t bytes = 0;
279 int64_t objsize;
280 int obj_order = 0;
281 char pool[RBD_MAX_POOL_NAME_SIZE];
282 char name[RBD_MAX_IMAGE_NAME_SIZE];
283 char snap_buf[RBD_MAX_SNAP_NAME_SIZE];
284 char conf[RBD_MAX_CONF_SIZE];
285 char clientname_buf[RBD_MAX_CONF_SIZE];
286 char *clientname;
287 rados_t cluster;
288 rados_ioctx_t io_ctx;
289 int ret;
290
291 if (qemu_rbd_parsename(filename, pool, sizeof(pool),
292 snap_buf, sizeof(snap_buf),
293 name, sizeof(name),
294 conf, sizeof(conf)) < 0) {
295 return -EINVAL;
296 }
297
298 /* Read out options */
299 while (options && options->name) {
300 if (!strcmp(options->name, BLOCK_OPT_SIZE)) {
301 bytes = options->value.n;
302 } else if (!strcmp(options->name, BLOCK_OPT_CLUSTER_SIZE)) {
303 if (options->value.n) {
304 objsize = options->value.n;
305 if ((objsize - 1) & objsize) { /* not a power of 2? */
306 error_report("obj size needs to be power of 2");
307 return -EINVAL;
308 }
309 if (objsize < 4096) {
310 error_report("obj size too small");
311 return -EINVAL;
312 }
313 obj_order = ffs(objsize) - 1;
314 }
315 }
316 options++;
317 }
318
319 clientname = qemu_rbd_parse_clientname(conf, clientname_buf);
320 if (rados_create(&cluster, clientname) < 0) {
321 error_report("error initializing");
322 return -EIO;
323 }
324
325 if (strstr(conf, "conf=") == NULL) {
326 /* try default location, but ignore failure */
327 rados_conf_read_file(cluster, NULL);
328 }
329
330 if (conf[0] != '\0' &&
331 qemu_rbd_set_conf(cluster, conf) < 0) {
332 error_report("error setting config options");
333 rados_shutdown(cluster);
334 return -EIO;
335 }
336
337 if (rados_connect(cluster) < 0) {
338 error_report("error connecting");
339 rados_shutdown(cluster);
340 return -EIO;
341 }
342
343 if (rados_ioctx_create(cluster, pool, &io_ctx) < 0) {
344 error_report("error opening pool %s", pool);
345 rados_shutdown(cluster);
346 return -EIO;
347 }
348
349 ret = rbd_create(io_ctx, name, bytes, &obj_order);
350 rados_ioctx_destroy(io_ctx);
351 rados_shutdown(cluster);
352
353 return ret;
354 }
355
356 /*
357 * This aio completion is being called from qemu_rbd_aio_event_reader()
358 * and runs in qemu context. It schedules a bh, but just in case the aio
359 * was not cancelled before.
360 */
361 static void qemu_rbd_complete_aio(RADOSCB *rcb)
362 {
363 RBDAIOCB *acb = rcb->acb;
364 int64_t r;
365
366 if (acb->cancelled) {
367 qemu_vfree(acb->bounce);
368 qemu_aio_release(acb);
369 goto done;
370 }
371
372 r = rcb->ret;
373
374 if (acb->write) {
375 if (r < 0) {
376 acb->ret = r;
377 acb->error = 1;
378 } else if (!acb->error) {
379 acb->ret = rcb->size;
380 }
381 } else {
382 if (r < 0) {
383 memset(rcb->buf, 0, rcb->size);
384 acb->ret = r;
385 acb->error = 1;
386 } else if (r < rcb->size) {
387 memset(rcb->buf + r, 0, rcb->size - r);
388 if (!acb->error) {
389 acb->ret = rcb->size;
390 }
391 } else if (!acb->error) {
392 acb->ret = r;
393 }
394 }
395 /* Note that acb->bh can be NULL in case where the aio was cancelled */
396 acb->bh = qemu_bh_new(rbd_aio_bh_cb, acb);
397 qemu_bh_schedule(acb->bh);
398 done:
399 g_free(rcb);
400 }
401
402 /*
403 * aio fd read handler. It runs in the qemu context and calls the
404 * completion handling of completed rados aio operations.
405 */
406 static void qemu_rbd_aio_event_reader(void *opaque)
407 {
408 BDRVRBDState *s = opaque;
409
410 ssize_t ret;
411
412 do {
413 char *p = (char *)&s->event_rcb;
414
415 /* now read the rcb pointer that was sent from a non qemu thread */
416 ret = read(s->fds[RBD_FD_READ], p + s->event_reader_pos,
417 sizeof(s->event_rcb) - s->event_reader_pos);
418 if (ret > 0) {
419 s->event_reader_pos += ret;
420 if (s->event_reader_pos == sizeof(s->event_rcb)) {
421 s->event_reader_pos = 0;
422 qemu_rbd_complete_aio(s->event_rcb);
423 s->qemu_aio_count--;
424 }
425 }
426 } while (ret < 0 && errno == EINTR);
427 }
428
429 static int qemu_rbd_aio_flush_cb(void *opaque)
430 {
431 BDRVRBDState *s = opaque;
432
433 return (s->qemu_aio_count > 0);
434 }
435
436 static int qemu_rbd_open(BlockDriverState *bs, const char *filename, int flags)
437 {
438 BDRVRBDState *s = bs->opaque;
439 char pool[RBD_MAX_POOL_NAME_SIZE];
440 char snap_buf[RBD_MAX_SNAP_NAME_SIZE];
441 char conf[RBD_MAX_CONF_SIZE];
442 char clientname_buf[RBD_MAX_CONF_SIZE];
443 char *clientname;
444 int r;
445
446 if (qemu_rbd_parsename(filename, pool, sizeof(pool),
447 snap_buf, sizeof(snap_buf),
448 s->name, sizeof(s->name),
449 conf, sizeof(conf)) < 0) {
450 return -EINVAL;
451 }
452
453 clientname = qemu_rbd_parse_clientname(conf, clientname_buf);
454 r = rados_create(&s->cluster, clientname);
455 if (r < 0) {
456 error_report("error initializing");
457 return r;
458 }
459
460 s->snap = NULL;
461 if (snap_buf[0] != '\0') {
462 s->snap = g_strdup(snap_buf);
463 }
464
465 if (strstr(conf, "conf=") == NULL) {
466 /* try default location, but ignore failure */
467 rados_conf_read_file(s->cluster, NULL);
468 }
469
470 if (conf[0] != '\0') {
471 r = qemu_rbd_set_conf(s->cluster, conf);
472 if (r < 0) {
473 error_report("error setting config options");
474 goto failed_shutdown;
475 }
476 }
477
478 r = rados_connect(s->cluster);
479 if (r < 0) {
480 error_report("error connecting");
481 goto failed_shutdown;
482 }
483
484 r = rados_ioctx_create(s->cluster, pool, &s->io_ctx);
485 if (r < 0) {
486 error_report("error opening pool %s", pool);
487 goto failed_shutdown;
488 }
489
490 r = rbd_open(s->io_ctx, s->name, &s->image, s->snap);
491 if (r < 0) {
492 error_report("error reading header from %s", s->name);
493 goto failed_open;
494 }
495
496 bs->read_only = (s->snap != NULL);
497
498 s->event_reader_pos = 0;
499 r = qemu_pipe(s->fds);
500 if (r < 0) {
501 error_report("error opening eventfd");
502 goto failed;
503 }
504 fcntl(s->fds[0], F_SETFL, O_NONBLOCK);
505 fcntl(s->fds[1], F_SETFL, O_NONBLOCK);
506 qemu_aio_set_fd_handler(s->fds[RBD_FD_READ], qemu_rbd_aio_event_reader,
507 NULL, qemu_rbd_aio_flush_cb, NULL, s);
508
509
510 return 0;
511
512 failed:
513 rbd_close(s->image);
514 failed_open:
515 rados_ioctx_destroy(s->io_ctx);
516 failed_shutdown:
517 rados_shutdown(s->cluster);
518 g_free(s->snap);
519 return r;
520 }
521
522 static void qemu_rbd_close(BlockDriverState *bs)
523 {
524 BDRVRBDState *s = bs->opaque;
525
526 close(s->fds[0]);
527 close(s->fds[1]);
528 qemu_aio_set_fd_handler(s->fds[RBD_FD_READ], NULL , NULL, NULL, NULL,
529 NULL);
530
531 rbd_close(s->image);
532 rados_ioctx_destroy(s->io_ctx);
533 g_free(s->snap);
534 rados_shutdown(s->cluster);
535 }
536
537 /*
538 * Cancel aio. Since we don't reference acb in a non qemu threads,
539 * it is safe to access it here.
540 */
541 static void qemu_rbd_aio_cancel(BlockDriverAIOCB *blockacb)
542 {
543 RBDAIOCB *acb = (RBDAIOCB *) blockacb;
544 acb->cancelled = 1;
545 }
546
547 static AIOPool rbd_aio_pool = {
548 .aiocb_size = sizeof(RBDAIOCB),
549 .cancel = qemu_rbd_aio_cancel,
550 };
551
552 static int qemu_rbd_send_pipe(BDRVRBDState *s, RADOSCB *rcb)
553 {
554 int ret = 0;
555 while (1) {
556 fd_set wfd;
557 int fd = s->fds[RBD_FD_WRITE];
558
559 /* send the op pointer to the qemu thread that is responsible
560 for the aio/op completion. Must do it in a qemu thread context */
561 ret = write(fd, (void *)&rcb, sizeof(rcb));
562 if (ret >= 0) {
563 break;
564 }
565 if (errno == EINTR) {
566 continue;
567 }
568 if (errno != EAGAIN) {
569 break;
570 }
571
572 FD_ZERO(&wfd);
573 FD_SET(fd, &wfd);
574 do {
575 ret = select(fd + 1, NULL, &wfd, NULL, NULL);
576 } while (ret < 0 && errno == EINTR);
577 }
578
579 return ret;
580 }
581
582 /*
583 * This is the callback function for rbd_aio_read and _write
584 *
585 * Note: this function is being called from a non qemu thread so
586 * we need to be careful about what we do here. Generally we only
587 * write to the block notification pipe, and do the rest of the
588 * io completion handling from qemu_rbd_aio_event_reader() which
589 * runs in a qemu context.
590 */
591 static void rbd_finish_aiocb(rbd_completion_t c, RADOSCB *rcb)
592 {
593 int ret;
594 rcb->ret = rbd_aio_get_return_value(c);
595 rbd_aio_release(c);
596 ret = qemu_rbd_send_pipe(rcb->s, rcb);
597 if (ret < 0) {
598 error_report("failed writing to acb->s->fds");
599 g_free(rcb);
600 }
601 }
602
603 /* Callback when all queued rbd_aio requests are complete */
604
605 static void rbd_aio_bh_cb(void *opaque)
606 {
607 RBDAIOCB *acb = opaque;
608
609 if (!acb->write) {
610 qemu_iovec_from_buffer(acb->qiov, acb->bounce, acb->qiov->size);
611 }
612 qemu_vfree(acb->bounce);
613 acb->common.cb(acb->common.opaque, (acb->ret > 0 ? 0 : acb->ret));
614 qemu_bh_delete(acb->bh);
615 acb->bh = NULL;
616
617 qemu_aio_release(acb);
618 }
619
620 static BlockDriverAIOCB *rbd_aio_rw_vector(BlockDriverState *bs,
621 int64_t sector_num,
622 QEMUIOVector *qiov,
623 int nb_sectors,
624 BlockDriverCompletionFunc *cb,
625 void *opaque, int write)
626 {
627 RBDAIOCB *acb;
628 RADOSCB *rcb;
629 rbd_completion_t c;
630 int64_t off, size;
631 char *buf;
632 int r;
633
634 BDRVRBDState *s = bs->opaque;
635
636 acb = qemu_aio_get(&rbd_aio_pool, bs, cb, opaque);
637 acb->write = write;
638 acb->qiov = qiov;
639 acb->bounce = qemu_blockalign(bs, qiov->size);
640 acb->ret = 0;
641 acb->error = 0;
642 acb->s = s;
643 acb->cancelled = 0;
644 acb->bh = NULL;
645
646 if (write) {
647 qemu_iovec_to_buffer(acb->qiov, acb->bounce);
648 }
649
650 buf = acb->bounce;
651
652 off = sector_num * BDRV_SECTOR_SIZE;
653 size = nb_sectors * BDRV_SECTOR_SIZE;
654
655 s->qemu_aio_count++; /* All the RADOSCB */
656
657 rcb = g_malloc(sizeof(RADOSCB));
658 rcb->done = 0;
659 rcb->acb = acb;
660 rcb->buf = buf;
661 rcb->s = acb->s;
662 rcb->size = size;
663 r = rbd_aio_create_completion(rcb, (rbd_callback_t) rbd_finish_aiocb, &c);
664 if (r < 0) {
665 goto failed;
666 }
667
668 if (write) {
669 r = rbd_aio_write(s->image, off, size, buf, c);
670 } else {
671 r = rbd_aio_read(s->image, off, size, buf, c);
672 }
673
674 if (r < 0) {
675 goto failed;
676 }
677
678 return &acb->common;
679
680 failed:
681 g_free(rcb);
682 s->qemu_aio_count--;
683 qemu_aio_release(acb);
684 return NULL;
685 }
686
687 static BlockDriverAIOCB *qemu_rbd_aio_readv(BlockDriverState *bs,
688 int64_t sector_num,
689 QEMUIOVector *qiov,
690 int nb_sectors,
691 BlockDriverCompletionFunc *cb,
692 void *opaque)
693 {
694 return rbd_aio_rw_vector(bs, sector_num, qiov, nb_sectors, cb, opaque, 0);
695 }
696
697 static BlockDriverAIOCB *qemu_rbd_aio_writev(BlockDriverState *bs,
698 int64_t sector_num,
699 QEMUIOVector *qiov,
700 int nb_sectors,
701 BlockDriverCompletionFunc *cb,
702 void *opaque)
703 {
704 return rbd_aio_rw_vector(bs, sector_num, qiov, nb_sectors, cb, opaque, 1);
705 }
706
707 static int qemu_rbd_co_flush(BlockDriverState *bs)
708 {
709 #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 1)
710 /* rbd_flush added in 0.1.1 */
711 BDRVRBDState *s = bs->opaque;
712 return rbd_flush(s->image);
713 #else
714 return 0;
715 #endif
716 }
717
718 static int qemu_rbd_getinfo(BlockDriverState *bs, BlockDriverInfo *bdi)
719 {
720 BDRVRBDState *s = bs->opaque;
721 rbd_image_info_t info;
722 int r;
723
724 r = rbd_stat(s->image, &info, sizeof(info));
725 if (r < 0) {
726 return r;
727 }
728
729 bdi->cluster_size = info.obj_size;
730 return 0;
731 }
732
733 static int64_t qemu_rbd_getlength(BlockDriverState *bs)
734 {
735 BDRVRBDState *s = bs->opaque;
736 rbd_image_info_t info;
737 int r;
738
739 r = rbd_stat(s->image, &info, sizeof(info));
740 if (r < 0) {
741 return r;
742 }
743
744 return info.size;
745 }
746
747 static int qemu_rbd_truncate(BlockDriverState *bs, int64_t offset)
748 {
749 BDRVRBDState *s = bs->opaque;
750 int r;
751
752 r = rbd_resize(s->image, offset);
753 if (r < 0) {
754 return r;
755 }
756
757 return 0;
758 }
759
760 static int qemu_rbd_snap_create(BlockDriverState *bs,
761 QEMUSnapshotInfo *sn_info)
762 {
763 BDRVRBDState *s = bs->opaque;
764 int r;
765
766 if (sn_info->name[0] == '\0') {
767 return -EINVAL; /* we need a name for rbd snapshots */
768 }
769
770 /*
771 * rbd snapshots are using the name as the user controlled unique identifier
772 * we can't use the rbd snapid for that purpose, as it can't be set
773 */
774 if (sn_info->id_str[0] != '\0' &&
775 strcmp(sn_info->id_str, sn_info->name) != 0) {
776 return -EINVAL;
777 }
778
779 if (strlen(sn_info->name) >= sizeof(sn_info->id_str)) {
780 return -ERANGE;
781 }
782
783 r = rbd_snap_create(s->image, sn_info->name);
784 if (r < 0) {
785 error_report("failed to create snap: %s", strerror(-r));
786 return r;
787 }
788
789 return 0;
790 }
791
792 static int qemu_rbd_snap_remove(BlockDriverState *bs,
793 const char *snapshot_name)
794 {
795 BDRVRBDState *s = bs->opaque;
796 int r;
797
798 r = rbd_snap_remove(s->image, snapshot_name);
799 return r;
800 }
801
802 static int qemu_rbd_snap_rollback(BlockDriverState *bs,
803 const char *snapshot_name)
804 {
805 BDRVRBDState *s = bs->opaque;
806 int r;
807
808 r = rbd_snap_rollback(s->image, snapshot_name);
809 return r;
810 }
811
812 static int qemu_rbd_snap_list(BlockDriverState *bs,
813 QEMUSnapshotInfo **psn_tab)
814 {
815 BDRVRBDState *s = bs->opaque;
816 QEMUSnapshotInfo *sn_info, *sn_tab = NULL;
817 int i, snap_count;
818 rbd_snap_info_t *snaps;
819 int max_snaps = RBD_MAX_SNAPS;
820
821 do {
822 snaps = g_malloc(sizeof(*snaps) * max_snaps);
823 snap_count = rbd_snap_list(s->image, snaps, &max_snaps);
824 if (snap_count < 0) {
825 g_free(snaps);
826 }
827 } while (snap_count == -ERANGE);
828
829 if (snap_count <= 0) {
830 goto done;
831 }
832
833 sn_tab = g_malloc0(snap_count * sizeof(QEMUSnapshotInfo));
834
835 for (i = 0; i < snap_count; i++) {
836 const char *snap_name = snaps[i].name;
837
838 sn_info = sn_tab + i;
839 pstrcpy(sn_info->id_str, sizeof(sn_info->id_str), snap_name);
840 pstrcpy(sn_info->name, sizeof(sn_info->name), snap_name);
841
842 sn_info->vm_state_size = snaps[i].size;
843 sn_info->date_sec = 0;
844 sn_info->date_nsec = 0;
845 sn_info->vm_clock_nsec = 0;
846 }
847 rbd_snap_list_end(snaps);
848
849 done:
850 *psn_tab = sn_tab;
851 return snap_count;
852 }
853
854 static QEMUOptionParameter qemu_rbd_create_options[] = {
855 {
856 .name = BLOCK_OPT_SIZE,
857 .type = OPT_SIZE,
858 .help = "Virtual disk size"
859 },
860 {
861 .name = BLOCK_OPT_CLUSTER_SIZE,
862 .type = OPT_SIZE,
863 .help = "RBD object size"
864 },
865 {NULL}
866 };
867
868 static BlockDriver bdrv_rbd = {
869 .format_name = "rbd",
870 .instance_size = sizeof(BDRVRBDState),
871 .bdrv_file_open = qemu_rbd_open,
872 .bdrv_close = qemu_rbd_close,
873 .bdrv_create = qemu_rbd_create,
874 .bdrv_get_info = qemu_rbd_getinfo,
875 .create_options = qemu_rbd_create_options,
876 .bdrv_getlength = qemu_rbd_getlength,
877 .bdrv_truncate = qemu_rbd_truncate,
878 .protocol_name = "rbd",
879
880 .bdrv_aio_readv = qemu_rbd_aio_readv,
881 .bdrv_aio_writev = qemu_rbd_aio_writev,
882 .bdrv_co_flush_to_disk = qemu_rbd_co_flush,
883
884 .bdrv_snapshot_create = qemu_rbd_snap_create,
885 .bdrv_snapshot_delete = qemu_rbd_snap_remove,
886 .bdrv_snapshot_list = qemu_rbd_snap_list,
887 .bdrv_snapshot_goto = qemu_rbd_snap_rollback,
888 };
889
890 static void bdrv_rbd_init(void)
891 {
892 bdrv_register(&bdrv_rbd);
893 }
894
895 block_init(bdrv_rbd_init);