![[BACK]](/cvsweb/icons/back.gif){kind=icon}
Return to rbd.c CVS log ![[TXT]](/cvsweb/icons/text.gif){kind=icon}
![[DIR]](/cvsweb/icons/dir.gif){kind=icon}
Up to [Qemu by Fabrice Bellard] / qemu / block|
Return to rbd.c CVS log | Up to [Qemu by Fabrice Bellard] / qemu / block |
1.1 root 1: /*
2: * QEMU Block driver for RADOS (Ceph)
3: *
4: * Copyright (C) 2010 Christian Brunner <[email protected]>
5: *
6: * This work is licensed under the terms of the GNU GPL, version 2. See
7: * the COPYING file in the top-level directory.
8: *
9: */
10:
11: #include "qemu-common.h"
12: #include "qemu-error.h"
13:
14: #include "rbd_types.h"
15: #include "block_int.h"
16:
17: #include <rados/librados.h>
18:
19:
20:
21: /*
22: * When specifying the image filename use:
23: *
24: * rbd:poolname/devicename
25: *
26: * poolname must be the name of an existing rados pool
27: *
28: * devicename is the basename for all objects used to
29: * emulate the raw device.
30: *
31: * Metadata information (image size, ...) is stored in an
32: * object with the name "devicename.rbd".
33: *
34: * The raw device is split into 4MB sized objects by default.
35: * The sequencenumber is encoded in a 12 byte long hex-string,
36: * and is attached to the devicename, separated by a dot.
37: * e.g. "devicename.1234567890ab"
38: *
39: */
40:
41: #define OBJ_MAX_SIZE (1UL << OBJ_DEFAULT_OBJ_ORDER)
42:
43: typedef struct RBDAIOCB {
44: BlockDriverAIOCB common;
45: QEMUBH *bh;
46: int ret;
47: QEMUIOVector *qiov;
48: char *bounce;
49: int write;
50: int64_t sector_num;
51: int aiocnt;
52: int error;
53: struct BDRVRBDState *s;
54: int cancelled;
55: } RBDAIOCB;
56:
57: typedef struct RADOSCB {
58: int rcbid;
59: RBDAIOCB *acb;
60: struct BDRVRBDState *s;
61: int done;
62: int64_t segsize;
63: char *buf;
64: int ret;
65: } RADOSCB;
66:
67: #define RBD_FD_READ 0
68: #define RBD_FD_WRITE 1
69:
70: typedef struct BDRVRBDState {
71: int fds[2];
72: rados_pool_t pool;
73: rados_pool_t header_pool;
74: char name[RBD_MAX_OBJ_NAME_SIZE];
75: char block_name[RBD_MAX_BLOCK_NAME_SIZE];
76: uint64_t size;
77: uint64_t objsize;
78: int qemu_aio_count;
79: int event_reader_pos;
80: RADOSCB *event_rcb;
81: } BDRVRBDState;
82:
83: typedef struct rbd_obj_header_ondisk RbdHeader1;
84:
85: static void rbd_aio_bh_cb(void *opaque);
86:
87: static int rbd_next_tok(char *dst, int dst_len,
88: char *src, char delim,
89: const char *name,
90: char **p)
91: {
92: int l;
93: char *end;
94:
95: *p = NULL;
96:
97: if (delim != '\0') {
98: end = strchr(src, delim);
99: if (end) {
100: *p = end + 1;
101: *end = '\0';
102: }
103: }
104: l = strlen(src);
105: if (l >= dst_len) {
106: error_report("%s too long", name);
107: return -EINVAL;
108: } else if (l == 0) {
109: error_report("%s too short", name);
110: return -EINVAL;
111: }
112:
113: pstrcpy(dst, dst_len, src);
114:
115: return 0;
116: }
117:
118: static int rbd_parsename(const char *filename,
119: char *pool, int pool_len,
120: char *snap, int snap_len,
121: char *name, int name_len)
122: {
123: const char *start;
124: char *p, *buf;
125: int ret;
126:
127: if (!strstart(filename, "rbd:", &start)) {
128: return -EINVAL;
129: }
130:
131: buf = qemu_strdup(start);
132: p = buf;
133:
134: ret = rbd_next_tok(pool, pool_len, p, '/', "pool name", &p);
135: if (ret < 0 || !p) {
136: ret = -EINVAL;
137: goto done;
138: }
139: ret = rbd_next_tok(name, name_len, p, '@', "object name", &p);
140: if (ret < 0) {
141: goto done;
142: }
143: if (!p) {
144: *snap = '\0';
145: goto done;
146: }
147:
148: ret = rbd_next_tok(snap, snap_len, p, '\0', "snap name", &p);
149:
150: done:
151: qemu_free(buf);
152: return ret;
153: }
154:
155: static int create_tmap_op(uint8_t op, const char *name, char **tmap_desc)
156: {
157: uint32_t len = strlen(name);
158: uint32_t len_le = cpu_to_le32(len);
159: /* total_len = encoding op + name + empty buffer */
160: uint32_t total_len = 1 + (sizeof(uint32_t) + len) + sizeof(uint32_t);
161: uint8_t *desc = NULL;
162:
163: desc = qemu_malloc(total_len);
164:
165: *tmap_desc = (char *)desc;
166:
167: *desc = op;
168: desc++;
169: memcpy(desc, &len_le, sizeof(len_le));
170: desc += sizeof(len_le);
171: memcpy(desc, name, len);
172: desc += len;
173: len = 0; /* no need for endian conversion for 0 */
174: memcpy(desc, &len, sizeof(len));
175: desc += sizeof(len);
176:
177: return (char *)desc - *tmap_desc;
178: }
179:
180: static void free_tmap_op(char *tmap_desc)
181: {
182: qemu_free(tmap_desc);
183: }
184:
185: static int rbd_register_image(rados_pool_t pool, const char *name)
186: {
187: char *tmap_desc;
188: const char *dir = RBD_DIRECTORY;
189: int ret;
190:
191: ret = create_tmap_op(CEPH_OSD_TMAP_SET, name, &tmap_desc);
192: if (ret < 0) {
193: return ret;
194: }
195:
196: ret = rados_tmap_update(pool, dir, tmap_desc, ret);
197: free_tmap_op(tmap_desc);
198:
199: return ret;
200: }
201:
202: static int touch_rbd_info(rados_pool_t pool, const char *info_oid)
203: {
204: int r = rados_write(pool, info_oid, 0, NULL, 0);
205: if (r < 0) {
206: return r;
207: }
208: return 0;
209: }
210:
211: static int rbd_assign_bid(rados_pool_t pool, uint64_t *id)
212: {
213: uint64_t out[1];
214: const char *info_oid = RBD_INFO;
215:
216: *id = 0;
217:
218: int r = touch_rbd_info(pool, info_oid);
219: if (r < 0) {
220: return r;
221: }
222:
223: r = rados_exec(pool, info_oid, "rbd", "assign_bid", NULL,
224: 0, (char *)out, sizeof(out));
225: if (r < 0) {
226: return r;
227: }
228:
229: le64_to_cpus(out);
230: *id = out[0];
231:
232: return 0;
233: }
234:
235: static int rbd_create(const char *filename, QEMUOptionParameter *options)
236: {
237: int64_t bytes = 0;
238: int64_t objsize;
239: uint64_t size;
240: time_t mtime;
241: uint8_t obj_order = RBD_DEFAULT_OBJ_ORDER;
242: char pool[RBD_MAX_SEG_NAME_SIZE];
243: char n[RBD_MAX_SEG_NAME_SIZE];
244: char name[RBD_MAX_OBJ_NAME_SIZE];
245: char snap_buf[RBD_MAX_SEG_NAME_SIZE];
246: char *snap = NULL;
247: RbdHeader1 header;
248: rados_pool_t p;
249: uint64_t bid;
250: uint32_t hi, lo;
251: int ret;
252:
253: if (rbd_parsename(filename,
254: pool, sizeof(pool),
255: snap_buf, sizeof(snap_buf),
256: name, sizeof(name)) < 0) {
257: return -EINVAL;
258: }
259: if (snap_buf[0] != '\0') {
260: snap = snap_buf;
261: }
262:
263: snprintf(n, sizeof(n), "%s%s", name, RBD_SUFFIX);
264:
265: /* Read out options */
266: while (options && options->name) {
267: if (!strcmp(options->name, BLOCK_OPT_SIZE)) {
268: bytes = options->value.n;
269: } else if (!strcmp(options->name, BLOCK_OPT_CLUSTER_SIZE)) {
270: if (options->value.n) {
271: objsize = options->value.n;
272: if ((objsize - 1) & objsize) { /* not a power of 2? */
273: error_report("obj size needs to be power of 2");
274: return -EINVAL;
275: }
276: if (objsize < 4096) {
277: error_report("obj size too small");
278: return -EINVAL;
279: }
280: obj_order = ffs(objsize) - 1;
281: }
282: }
283: options++;
284: }
285:
286: memset(&header, 0, sizeof(header));
287: pstrcpy(header.text, sizeof(header.text), RBD_HEADER_TEXT);
288: pstrcpy(header.signature, sizeof(header.signature), RBD_HEADER_SIGNATURE);
289: pstrcpy(header.version, sizeof(header.version), RBD_HEADER_VERSION);
290: header.image_size = cpu_to_le64(bytes);
291: header.options.order = obj_order;
292: header.options.crypt_type = RBD_CRYPT_NONE;
293: header.options.comp_type = RBD_COMP_NONE;
294: header.snap_seq = 0;
295: header.snap_count = 0;
296:
297: if (rados_initialize(0, NULL) < 0) {
298: error_report("error initializing");
299: return -EIO;
300: }
301:
302: if (rados_open_pool(pool, &p)) {
303: error_report("error opening pool %s", pool);
304: rados_deinitialize();
305: return -EIO;
306: }
307:
308: /* check for existing rbd header file */
309: ret = rados_stat(p, n, &size, &mtime);
310: if (ret == 0) {
311: ret=-EEXIST;
312: goto done;
313: }
314:
315: ret = rbd_assign_bid(p, &bid);
316: if (ret < 0) {
317: error_report("failed assigning block id");
318: rados_deinitialize();
319: return -EIO;
320: }
321: hi = bid >> 32;
322: lo = bid & 0xFFFFFFFF;
323: snprintf(header.block_name, sizeof(header.block_name), "rb.%x.%x", hi, lo);
324:
325: /* create header file */
326: ret = rados_write(p, n, 0, (const char *)&header, sizeof(header));
327: if (ret < 0) {
328: goto done;
329: }
330:
331: ret = rbd_register_image(p, name);
332: done:
333: rados_close_pool(p);
334: rados_deinitialize();
335:
336: return ret;
337: }
338:
339: /*
340: * This aio completion is being called from rbd_aio_event_reader() and
341: * runs in qemu context. It schedules a bh, but just in case the aio
342: * was not cancelled before.
343: */
344: static void rbd_complete_aio(RADOSCB *rcb)
345: {
346: RBDAIOCB *acb = rcb->acb;
347: int64_t r;
348:
349: acb->aiocnt--;
350:
351: if (acb->cancelled) {
352: if (!acb->aiocnt) {
353: qemu_vfree(acb->bounce);
354: qemu_aio_release(acb);
355: }
356: goto done;
357: }
358:
359: r = rcb->ret;
360:
361: if (acb->write) {
362: if (r < 0) {
363: acb->ret = r;
364: acb->error = 1;
365: } else if (!acb->error) {
366: acb->ret += rcb->segsize;
367: }
368: } else {
369: if (r == -ENOENT) {
370: memset(rcb->buf, 0, rcb->segsize);
371: if (!acb->error) {
372: acb->ret += rcb->segsize;
373: }
374: } else if (r < 0) {
375: memset(rcb->buf, 0, rcb->segsize);
376: acb->ret = r;
377: acb->error = 1;
378: } else if (r < rcb->segsize) {
379: memset(rcb->buf + r, 0, rcb->segsize - r);
380: if (!acb->error) {
381: acb->ret += rcb->segsize;
382: }
383: } else if (!acb->error) {
384: acb->ret += r;
385: }
386: }
387: /* Note that acb->bh can be NULL in case where the aio was cancelled */
388: if (!acb->aiocnt) {
389: acb->bh = qemu_bh_new(rbd_aio_bh_cb, acb);
390: qemu_bh_schedule(acb->bh);
391: }
392: done:
393: qemu_free(rcb);
394: }
395:
396: /*
397: * aio fd read handler. It runs in the qemu context and calls the
398: * completion handling of completed rados aio operations.
399: */
400: static void rbd_aio_event_reader(void *opaque)
401: {
402: BDRVRBDState *s = opaque;
403:
404: ssize_t ret;
405:
406: do {
407: char *p = (char *)&s->event_rcb;
408:
409: /* now read the rcb pointer that was sent from a non qemu thread */
410: if ((ret = read(s->fds[RBD_FD_READ], p + s->event_reader_pos,
411: sizeof(s->event_rcb) - s->event_reader_pos)) > 0) {
412: if (ret > 0) {
413: s->event_reader_pos += ret;
414: if (s->event_reader_pos == sizeof(s->event_rcb)) {
415: s->event_reader_pos = 0;
416: rbd_complete_aio(s->event_rcb);
417: s->qemu_aio_count --;
418: }
419: }
420: }
421: } while (ret < 0 && errno == EINTR);
422: }
423:
424: static int rbd_aio_flush_cb(void *opaque)
425: {
426: BDRVRBDState *s = opaque;
427:
428: return (s->qemu_aio_count > 0);
429: }
430:
431:
432: static int rbd_set_snapc(rados_pool_t pool, const char *snap, RbdHeader1 *header)
433: {
434: uint32_t snap_count = le32_to_cpu(header->snap_count);
435: rados_snap_t *snaps = NULL;
436: rados_snap_t seq;
437: uint32_t i;
438: uint64_t snap_names_len = le64_to_cpu(header->snap_names_len);
439: int r;
440: rados_snap_t snapid = 0;
441:
442: if (snap_count) {
443: const char *header_snap = (const char *)&header->snaps[snap_count];
444: const char *end = header_snap + snap_names_len;
445: snaps = qemu_malloc(sizeof(rados_snap_t) * header->snap_count);
446:
447: for (i=0; i < snap_count; i++) {
448: snaps[i] = le64_to_cpu(header->snaps[i].id);
449:
450: if (snap && strcmp(snap, header_snap) == 0) {
451: snapid = snaps[i];
452: }
453:
454: header_snap += strlen(header_snap) + 1;
455: if (header_snap > end) {
456: error_report("bad header, snapshot list broken");
457: }
458: }
459: }
460:
461: if (snap && !snapid) {
462: error_report("snapshot not found");
463: qemu_free(snaps);
464: return -ENOENT;
465: }
466: seq = le32_to_cpu(header->snap_seq);
467:
468: r = rados_set_snap_context(pool, seq, snaps, snap_count);
469:
470: rados_set_snap(pool, snapid);
471:
472: qemu_free(snaps);
473:
474: return r;
475: }
476:
477: #define BUF_READ_START_LEN 4096
478:
479: static int rbd_read_header(BDRVRBDState *s, char **hbuf)
480: {
481: char *buf = NULL;
482: char n[RBD_MAX_SEG_NAME_SIZE];
483: uint64_t len = BUF_READ_START_LEN;
484: int r;
485:
486: snprintf(n, sizeof(n), "%s%s", s->name, RBD_SUFFIX);
487:
488: buf = qemu_malloc(len);
489:
490: r = rados_read(s->header_pool, n, 0, buf, len);
491: if (r < 0) {
492: goto failed;
493: }
494:
495: if (r < len) {
496: goto done;
497: }
498:
499: qemu_free(buf);
500: buf = qemu_malloc(len);
501:
502: r = rados_stat(s->header_pool, n, &len, NULL);
503: if (r < 0) {
504: goto failed;
505: }
506:
507: r = rados_read(s->header_pool, n, 0, buf, len);
508: if (r < 0) {
509: goto failed;
510: }
511:
512: done:
513: *hbuf = buf;
514: return 0;
515:
516: failed:
517: qemu_free(buf);
518: return r;
519: }
520:
521: static int rbd_open(BlockDriverState *bs, const char *filename, int flags)
522: {
523: BDRVRBDState *s = bs->opaque;
524: RbdHeader1 *header;
525: char pool[RBD_MAX_SEG_NAME_SIZE];
526: char snap_buf[RBD_MAX_SEG_NAME_SIZE];
527: char *snap = NULL;
528: char *hbuf = NULL;
529: int r;
530:
531: if (rbd_parsename(filename, pool, sizeof(pool),
532: snap_buf, sizeof(snap_buf),
533: s->name, sizeof(s->name)) < 0) {
534: return -EINVAL;
535: }
536: if (snap_buf[0] != '\0') {
537: snap = snap_buf;
538: }
539:
540: if ((r = rados_initialize(0, NULL)) < 0) {
541: error_report("error initializing");
542: return r;
543: }
544:
545: if ((r = rados_open_pool(pool, &s->pool))) {
546: error_report("error opening pool %s", pool);
547: rados_deinitialize();
548: return r;
549: }
550:
551: if ((r = rados_open_pool(pool, &s->header_pool))) {
552: error_report("error opening pool %s", pool);
553: rados_deinitialize();
554: return r;
555: }
556:
557: if ((r = rbd_read_header(s, &hbuf)) < 0) {
558: error_report("error reading header from %s", s->name);
559: goto failed;
560: }
561:
562: if (memcmp(hbuf + 64, RBD_HEADER_SIGNATURE, 4)) {
563: error_report("Invalid header signature");
564: r = -EMEDIUMTYPE;
565: goto failed;
566: }
567:
568: if (memcmp(hbuf + 68, RBD_HEADER_VERSION, 8)) {
569: error_report("Unknown image version");
570: r = -EMEDIUMTYPE;
571: goto failed;
572: }
573:
574: header = (RbdHeader1 *) hbuf;
575: s->size = le64_to_cpu(header->image_size);
576: s->objsize = 1ULL << header->options.order;
577: memcpy(s->block_name, header->block_name, sizeof(header->block_name));
578:
579: r = rbd_set_snapc(s->pool, snap, header);
580: if (r < 0) {
581: error_report("failed setting snap context: %s", strerror(-r));
582: goto failed;
583: }
584:
585: bs->read_only = (snap != NULL);
586:
587: s->event_reader_pos = 0;
588: r = qemu_pipe(s->fds);
589: if (r < 0) {
590: error_report("error opening eventfd");
591: goto failed;
592: }
593: fcntl(s->fds[0], F_SETFL, O_NONBLOCK);
594: fcntl(s->fds[1], F_SETFL, O_NONBLOCK);
595: qemu_aio_set_fd_handler(s->fds[RBD_FD_READ], rbd_aio_event_reader, NULL,
596: rbd_aio_flush_cb, NULL, s);
597:
598: qemu_free(hbuf);
599:
600: return 0;
601:
602: failed:
603: qemu_free(hbuf);
604:
605: rados_close_pool(s->header_pool);
606: rados_close_pool(s->pool);
607: rados_deinitialize();
608: return r;
609: }
610:
611: static void rbd_close(BlockDriverState *bs)
612: {
613: BDRVRBDState *s = bs->opaque;
614:
615: close(s->fds[0]);
616: close(s->fds[1]);
617: qemu_aio_set_fd_handler(s->fds[RBD_FD_READ], NULL , NULL, NULL, NULL,
618: NULL);
619:
620: rados_close_pool(s->header_pool);
621: rados_close_pool(s->pool);
622: rados_deinitialize();
623: }
624:
625: /*
626: * Cancel aio. Since we don't reference acb in a non qemu threads,
627: * it is safe to access it here.
628: */
629: static void rbd_aio_cancel(BlockDriverAIOCB *blockacb)
630: {
631: RBDAIOCB *acb = (RBDAIOCB *) blockacb;
632: acb->cancelled = 1;
633: }
634:
635: static AIOPool rbd_aio_pool = {
636: .aiocb_size = sizeof(RBDAIOCB),
637: .cancel = rbd_aio_cancel,
638: };
639:
640: /*
641: * This is the callback function for rados_aio_read and _write
642: *
643: * Note: this function is being called from a non qemu thread so
644: * we need to be careful about what we do here. Generally we only
645: * write to the block notification pipe, and do the rest of the
646: * io completion handling from rbd_aio_event_reader() which
647: * runs in a qemu context.
648: */
649: static void rbd_finish_aiocb(rados_completion_t c, RADOSCB *rcb)
650: {
651: int ret;
652: rcb->ret = rados_aio_get_return_value(c);
653: rados_aio_release(c);
654: while (1) {
655: fd_set wfd;
656: int fd = rcb->s->fds[RBD_FD_WRITE];
657:
658: /* send the rcb pointer to the qemu thread that is responsible
659: for the aio completion. Must do it in a qemu thread context */
660: ret = write(fd, (void *)&rcb, sizeof(rcb));
661: if (ret >= 0) {
662: break;
663: }
664: if (errno == EINTR) {
665: continue;
666: }
667: if (errno != EAGAIN) {
668: break;
669: }
670:
671: FD_ZERO(&wfd);
672: FD_SET(fd, &wfd);
673: do {
674: ret = select(fd + 1, NULL, &wfd, NULL, NULL);
675: } while (ret < 0 && errno == EINTR);
676: }
677:
678: if (ret < 0) {
679: error_report("failed writing to acb->s->fds\n");
680: qemu_free(rcb);
681: }
682: }
683:
684: /* Callback when all queued rados_aio requests are complete */
685:
686: static void rbd_aio_bh_cb(void *opaque)
687: {
688: RBDAIOCB *acb = opaque;
689:
690: if (!acb->write) {
691: qemu_iovec_from_buffer(acb->qiov, acb->bounce, acb->qiov->size);
692: }
693: qemu_vfree(acb->bounce);
694: acb->common.cb(acb->common.opaque, (acb->ret > 0 ? 0 : acb->ret));
695: qemu_bh_delete(acb->bh);
696: acb->bh = NULL;
697:
698: qemu_aio_release(acb);
699: }
700:
701: static BlockDriverAIOCB *rbd_aio_rw_vector(BlockDriverState *bs,
702: int64_t sector_num,
703: QEMUIOVector *qiov,
704: int nb_sectors,
705: BlockDriverCompletionFunc *cb,
706: void *opaque, int write)
707: {
708: RBDAIOCB *acb;
709: RADOSCB *rcb;
710: rados_completion_t c;
711: char n[RBD_MAX_SEG_NAME_SIZE];
712: int64_t segnr, segoffs, segsize, last_segnr;
713: int64_t off, size;
714: char *buf;
715:
716: BDRVRBDState *s = bs->opaque;
717:
718: acb = qemu_aio_get(&rbd_aio_pool, bs, cb, opaque);
719: acb->write = write;
720: acb->qiov = qiov;
721: acb->bounce = qemu_blockalign(bs, qiov->size);
722: acb->aiocnt = 0;
723: acb->ret = 0;
724: acb->error = 0;
725: acb->s = s;
726: acb->cancelled = 0;
727: acb->bh = NULL;
728:
729: if (write) {
730: qemu_iovec_to_buffer(acb->qiov, acb->bounce);
731: }
732:
733: buf = acb->bounce;
734:
735: off = sector_num * BDRV_SECTOR_SIZE;
736: size = nb_sectors * BDRV_SECTOR_SIZE;
737: segnr = off / s->objsize;
738: segoffs = off % s->objsize;
739: segsize = s->objsize - segoffs;
740:
741: last_segnr = ((off + size - 1) / s->objsize);
742: acb->aiocnt = (last_segnr - segnr) + 1;
743:
744: s->qemu_aio_count += acb->aiocnt; /* All the RADOSCB */
745:
746: while (size > 0) {
747: if (size < segsize) {
748: segsize = size;
749: }
750:
751: snprintf(n, sizeof(n), "%s.%012" PRIx64, s->block_name,
752: segnr);
753:
754: rcb = qemu_malloc(sizeof(RADOSCB));
755: rcb->done = 0;
756: rcb->acb = acb;
757: rcb->segsize = segsize;
758: rcb->buf = buf;
759: rcb->s = acb->s;
760:
761: if (write) {
762: rados_aio_create_completion(rcb, NULL,
763: (rados_callback_t) rbd_finish_aiocb,
764: &c);
765: rados_aio_write(s->pool, n, segoffs, buf, segsize, c);
766: } else {
767: rados_aio_create_completion(rcb,
768: (rados_callback_t) rbd_finish_aiocb,
769: NULL, &c);
770: rados_aio_read(s->pool, n, segoffs, buf, segsize, c);
771: }
772:
773: buf += segsize;
774: size -= segsize;
775: segoffs = 0;
776: segsize = s->objsize;
777: segnr++;
778: }
779:
780: return &acb->common;
781: }
782:
783: static BlockDriverAIOCB *rbd_aio_readv(BlockDriverState * bs,
784: int64_t sector_num, QEMUIOVector * qiov,
785: int nb_sectors,
786: BlockDriverCompletionFunc * cb,
787: void *opaque)
788: {
789: return rbd_aio_rw_vector(bs, sector_num, qiov, nb_sectors, cb, opaque, 0);
790: }
791:
792: static BlockDriverAIOCB *rbd_aio_writev(BlockDriverState * bs,
793: int64_t sector_num, QEMUIOVector * qiov,
794: int nb_sectors,
795: BlockDriverCompletionFunc * cb,
796: void *opaque)
797: {
798: return rbd_aio_rw_vector(bs, sector_num, qiov, nb_sectors, cb, opaque, 1);
799: }
800:
801: static int rbd_getinfo(BlockDriverState * bs, BlockDriverInfo * bdi)
802: {
803: BDRVRBDState *s = bs->opaque;
804: bdi->cluster_size = s->objsize;
805: return 0;
806: }
807:
808: static int64_t rbd_getlength(BlockDriverState * bs)
809: {
810: BDRVRBDState *s = bs->opaque;
811:
812: return s->size;
813: }
814:
815: static int rbd_snap_create(BlockDriverState *bs, QEMUSnapshotInfo *sn_info)
816: {
817: BDRVRBDState *s = bs->opaque;
818: char inbuf[512], outbuf[128];
819: uint64_t snap_id;
820: int r;
821: char *p = inbuf;
822: char *end = inbuf + sizeof(inbuf);
823: char n[RBD_MAX_SEG_NAME_SIZE];
824: char *hbuf = NULL;
825: RbdHeader1 *header;
826:
827: if (sn_info->name[0] == '\0') {
828: return -EINVAL; /* we need a name for rbd snapshots */
829: }
830:
831: /*
832: * rbd snapshots are using the name as the user controlled unique identifier
833: * we can't use the rbd snapid for that purpose, as it can't be set
834: */
835: if (sn_info->id_str[0] != '\0' &&
836: strcmp(sn_info->id_str, sn_info->name) != 0) {
837: return -EINVAL;
838: }
839:
840: if (strlen(sn_info->name) >= sizeof(sn_info->id_str)) {
841: return -ERANGE;
842: }
843:
844: r = rados_selfmanaged_snap_create(s->header_pool, &snap_id);
845: if (r < 0) {
846: error_report("failed to create snap id: %s", strerror(-r));
847: return r;
848: }
849:
850: *(uint32_t *)p = strlen(sn_info->name);
851: cpu_to_le32s((uint32_t *)p);
852: p += sizeof(uint32_t);
853: strncpy(p, sn_info->name, end - p);
854: p += strlen(p);
855: if (p + sizeof(snap_id) > end) {
856: error_report("invalid input parameter");
857: return -EINVAL;
858: }
859:
860: *(uint64_t *)p = snap_id;
861: cpu_to_le64s((uint64_t *)p);
862:
863: snprintf(n, sizeof(n), "%s%s", s->name, RBD_SUFFIX);
864:
865: r = rados_exec(s->header_pool, n, "rbd", "snap_add", inbuf,
866: sizeof(inbuf), outbuf, sizeof(outbuf));
867: if (r < 0) {
868: error_report("rbd.snap_add execution failed failed: %s", strerror(-r));
869: return r;
870: }
871:
872: sprintf(sn_info->id_str, "%s", sn_info->name);
873:
874: r = rbd_read_header(s, &hbuf);
875: if (r < 0) {
876: error_report("failed reading header: %s", strerror(-r));
877: return r;
878: }
879:
880: header = (RbdHeader1 *) hbuf;
881: r = rbd_set_snapc(s->pool, sn_info->name, header);
882: if (r < 0) {
883: error_report("failed setting snap context: %s", strerror(-r));
884: goto failed;
885: }
886:
887: return 0;
888:
889: failed:
890: qemu_free(header);
891: return r;
892: }
893:
894: static int decode32(char **p, const char *end, uint32_t *v)
895: {
896: if (*p + 4 > end) {
897: return -ERANGE;
898: }
899:
900: *v = *(uint32_t *)(*p);
901: le32_to_cpus(v);
902: *p += 4;
903: return 0;
904: }
905:
906: static int decode64(char **p, const char *end, uint64_t *v)
907: {
908: if (*p + 8 > end) {
909: return -ERANGE;
910: }
911:
912: *v = *(uint64_t *)(*p);
913: le64_to_cpus(v);
914: *p += 8;
915: return 0;
916: }
917:
918: static int decode_str(char **p, const char *end, char **s)
919: {
920: uint32_t len;
921: int r;
922:
923: if ((r = decode32(p, end, &len)) < 0) {
924: return r;
925: }
926:
927: *s = qemu_malloc(len + 1);
928: memcpy(*s, *p, len);
929: *p += len;
930: (*s)[len] = '\0';
931:
932: return len;
933: }
934:
935: static int rbd_snap_list(BlockDriverState *bs, QEMUSnapshotInfo **psn_tab)
936: {
937: BDRVRBDState *s = bs->opaque;
938: char n[RBD_MAX_SEG_NAME_SIZE];
939: QEMUSnapshotInfo *sn_info, *sn_tab = NULL;
940: RbdHeader1 *header;
941: char *hbuf = NULL;
942: char *outbuf = NULL, *end, *buf;
943: uint64_t len;
944: uint64_t snap_seq;
945: uint32_t snap_count;
946: int r, i;
947:
948: /* read header to estimate how much space we need to read the snap
949: * list */
950: if ((r = rbd_read_header(s, &hbuf)) < 0) {
951: goto done_err;
952: }
953: header = (RbdHeader1 *)hbuf;
954: len = le64_to_cpu(header->snap_names_len);
955: len += 1024; /* should have already been enough, but new snapshots might
956: already been created since we read the header. just allocate
957: a bit more, so that in most cases it'll suffice anyway */
958: qemu_free(hbuf);
959:
960: snprintf(n, sizeof(n), "%s%s", s->name, RBD_SUFFIX);
961: while (1) {
962: qemu_free(outbuf);
963: outbuf = qemu_malloc(len);
964:
965: r = rados_exec(s->header_pool, n, "rbd", "snap_list", NULL, 0,
966: outbuf, len);
967: if (r < 0) {
968: error_report("rbd.snap_list execution failed failed: %s", strerror(-r));
969: goto done_err;
970: }
971: if (r != len) {
972: break;
973: }
974:
975: /* if we're here, we probably raced with some snaps creation */
976: len *= 2;
977: }
978: buf = outbuf;
979: end = buf + len;
980:
981: if ((r = decode64(&buf, end, &snap_seq)) < 0) {
982: goto done_err;
983: }
984: if ((r = decode32(&buf, end, &snap_count)) < 0) {
985: goto done_err;
986: }
987:
988: sn_tab = qemu_mallocz(snap_count * sizeof(QEMUSnapshotInfo));
989: for (i = 0; i < snap_count; i++) {
990: uint64_t id, image_size;
991: char *snap_name;
992:
993: if ((r = decode64(&buf, end, &id)) < 0) {
994: goto done_err;
995: }
996: if ((r = decode64(&buf, end, &image_size)) < 0) {
997: goto done_err;
998: }
999: if ((r = decode_str(&buf, end, &snap_name)) < 0) {
1000: goto done_err;
1001: }
1002:
1003: sn_info = sn_tab + i;
1004: pstrcpy(sn_info->id_str, sizeof(sn_info->id_str), snap_name);
1005: pstrcpy(sn_info->name, sizeof(sn_info->name), snap_name);
1006: qemu_free(snap_name);
1007:
1008: sn_info->vm_state_size = image_size;
1009: sn_info->date_sec = 0;
1010: sn_info->date_nsec = 0;
1011: sn_info->vm_clock_nsec = 0;
1012: }
1013: *psn_tab = sn_tab;
1014: qemu_free(outbuf);
1015: return snap_count;
1016: done_err:
1017: qemu_free(sn_tab);
1018: qemu_free(outbuf);
1019: return r;
1020: }
1021:
1022: static QEMUOptionParameter rbd_create_options[] = {
1023: {
1024: .name = BLOCK_OPT_SIZE,
1025: .type = OPT_SIZE,
1026: .help = "Virtual disk size"
1027: },
1028: {
1029: .name = BLOCK_OPT_CLUSTER_SIZE,
1030: .type = OPT_SIZE,
1031: .help = "RBD object size"
1032: },
1033: {NULL}
1034: };
1035:
1036: static BlockDriver bdrv_rbd = {
1037: .format_name = "rbd",
1038: .instance_size = sizeof(BDRVRBDState),
1039: .bdrv_file_open = rbd_open,
1040: .bdrv_close = rbd_close,
1041: .bdrv_create = rbd_create,
1042: .bdrv_get_info = rbd_getinfo,
1043: .create_options = rbd_create_options,
1044: .bdrv_getlength = rbd_getlength,
1045: .protocol_name = "rbd",
1046:
1047: .bdrv_aio_readv = rbd_aio_readv,
1048: .bdrv_aio_writev = rbd_aio_writev,
1049:
1050: .bdrv_snapshot_create = rbd_snap_create,
1051: .bdrv_snapshot_list = rbd_snap_list,
1052: };
1053:
1054: static void bdrv_rbd_init(void)
1055: {
1056: bdrv_register(&bdrv_rbd);
1057: }
1058:
1059: block_init(bdrv_rbd_init);
This archive runs on limited infrastructure. Preserving old code on modern bandwidth. Automated agents are requested to crawl responsibly.