| Message ID | 20100527191158.GB16854@chb-desktop |
|---|---|
| State | New |
| Headers | show |
Am 27.05.2010 21:11, schrieb Christian Brunner: > This is a block driver for the distributed file system Ceph > (http://ceph.newdream.net/). This driver uses librados (which > is part of the Ceph server) for direct access to the Ceph object > store and is running entirely in userspace. Therefore it is > called "rbd" - rados block device. > > To compile the driver a recent version of ceph (unstable/testin git > head or 0.20.3 once it is released) is needed and you have to > "--enable-rbd" when running configure. > > Additional information is available on the Ceph-Wiki: > > http://ceph.newdream.net/wiki/Kvm-rbd > > The patch is based on git://repo.or.cz/qemu/kevin.git block Signed-off-by line is missing. > --- > Makefile | 3 + > Makefile.objs | 1 + > block/rbd.c | 584 +++++++++++++++++++++++++++++++++++++++++++++++++++++ > block/rbd_types.h | 52 +++++ > configure | 27 +++ > 5 files changed, 667 insertions(+), 0 deletions(-) > create mode 100644 block/rbd.c > create mode 100644 block/rbd_types.h > > diff --git a/Makefile b/Makefile > index 7986bf6..8d09612 100644 > --- a/Makefile > +++ b/Makefile > @@ -27,6 +27,9 @@ configure: ; > $(call set-vpath, $(SRC_PATH):$(SRC_PATH)/hw) > > LIBS+=-lz $(LIBS_TOOLS) > +ifdef CONFIG_RBD > +LIBS+=-lrados > +endif You already write the -lrados option to config-host.mak in configure, so this looks unnecessary. > > ifdef BUILD_DOCS > DOCS=qemu-doc.html qemu-tech.html qemu.1 qemu-img.1 qemu-nbd.8 > diff --git a/Makefile.objs b/Makefile.objs > index 1a942e5..08dc11f 100644 > --- a/Makefile.objs > +++ b/Makefile.objs > @@ -18,6 +18,7 @@ block-nested-y += parallels.o nbd.o blkdebug.o > block-nested-$(CONFIG_WIN32) += raw-win32.o > block-nested-$(CONFIG_POSIX) += raw-posix.o > block-nested-$(CONFIG_CURL) += curl.o > +block-nested-$(CONFIG_RBD) += rbd.o > > block-obj-y += $(addprefix block/, $(block-nested-y)) > > diff --git a/block/rbd.c b/block/rbd.c > new file mode 100644 > index 0000000..375ae9d > --- /dev/null > +++ b/block/rbd.c > @@ -0,0 +1,584 @@ > +/* > + * QEMU Block driver for RADOS (Ceph) > + * > + * Copyright (C) 2010 Christian Brunner <chb@muc.de> > + * > + * This work is licensed under the terms of the GNU GPL, version 2. See > + * the COPYING file in the top-level directory. > + * > + */ > + > +#include "qemu-common.h" > +#include <sys/types.h> > +#include <stdbool.h> > + > +#include <qemu-common.h> > + > +#include "rbd_types.h" > +#include "module.h" > +#include "block_int.h" > + > +#include <stdio.h> > +#include <stdlib.h> > +#include <rados/librados.h> > + > +#include <signal.h> > + > +/* > + * When specifying the image filename use: > + * > + * rbd:poolname/devicename > + * > + * poolname must be the name of an existing rados pool > + * > + * devicename is the basename for all objects used to > + * emulate the raw device. > + * > + * Metadata information (image size, ...) is stored in an > + * object with the name "devicename.rbd". > + * > + * The raw device is split into 4MB sized objects by default. > + * The sequencenumber is encoded in a 12 byte long hex-string, > + * and is attached to the devicename, separated by a dot. > + * e.g. "devicename.1234567890ab" > + * > + */ > + > +#define OBJ_MAX_SIZE (1UL << OBJ_DEFAULT_OBJ_ORDER) > + > +typedef struct RBDAIOCB { > + BlockDriverAIOCB common; > + QEMUBH *bh; > + int ret; > + QEMUIOVector *qiov; > + char *bounce; > + int write; > + int64_t sector_num; > + int aiocnt; > + int error; > +} RBDAIOCB; > + > +typedef struct RADOSCB { > + int rcbid; > + RBDAIOCB *acb; > + int done; > + int64_t segsize; > + char *buf; > +} RADOSCB; > + > +typedef struct RBDRVRBDState { > + rados_pool_t pool; > + char name[RBD_MAX_OBJ_NAME_SIZE]; > + int name_len; name_len looks unused. > + uint64_t size; > + uint64_t objsize; > +} RBDRVRBDState; Hm, you mean BDRVRBDState? Maybe ceph would have been a better driver name to avoid such type names. ;-) > + > +typedef struct rbd_obj_header_ondisk RbdHeader1; > + > +static int rbd_parsename(const char *filename, char *pool, char *name) > +{ > + const char *rbdname; > + char *p, *n; > + int l; > + > + if (!strstart(filename, "rbd:", &rbdname)) { > + return -EINVAL; > + } > + > + pstrcpy(pool, 2 * RBD_MAX_SEG_NAME_SIZE, rbdname); Why twice the size? The callers pass a char[RBD_MAX_SEG_NAME_SIZE], so doesn't this allow buffer overflows? > + p = strchr(pool, '/'); > + if (p == NULL) { > + return -EINVAL; > + } > + > + *p = '\0'; > + n = ++p; Why introduce a new variable here? p isn't used any more afterwards. > + > + l = strlen(n); > + > + if (l > RBD_MAX_OBJ_NAME_SIZE) { > + fprintf(stderr, "object name to long\n"); Off by one, you need to consider the trailing '\0'. Also, please use error_report instead of fprintf(stderr, ...) for real error messages. Directly printing to stderr is okay for debug code. > + return -EINVAL; > + } else if (l <= 0) { > + fprintf(stderr, "object name to short\n"); > + return -EINVAL; > + } > + > + strcpy(name, n); > + > + return l; > +} > + > +static int create_tmap_op(uint8_t op, const char *name, char **tmap_desc) > +{ > + uint32_t len = strlen(name); > + uint32_t total_len = 1 + (sizeof(uint32_t) + len) + sizeof(uint32_t); /* encoding op + name + empty buffer */ This is more than 80 characters. > + char *desc; > + > + desc = qemu_malloc(total_len); > + if (!desc) { > + return -ENOMEM; > + } qemu_malloc never returns NULL. > + > + *tmap_desc = desc; > + > + *desc = op; > + desc++; > + memcpy(desc, &len, sizeof(len)); > + desc += sizeof(len); > + memcpy(desc, name, len); > + desc += len; > + len = 0; > + memcpy(desc, &len, sizeof(len)); > + desc += sizeof(len); > + > + return desc - *tmap_desc; > +} > + > +static void free_tmap_op(char *tmap_desc) > +{ > + qemu_free(tmap_desc); > +} > + > +static int rbd_register_image(rados_pool_t pool, const char *name) > +{ > + char *tmap_desc; > + const char *dir = RBD_DIRECTORY; > + int ret; > + > + ret = create_tmap_op(CEPH_OSD_TMAP_SET, name, &tmap_desc); > + if (ret < 0) { > + return ret; > + } > + > + ret = rados_tmap_update(pool, dir, tmap_desc, ret); > + free_tmap_op(tmap_desc); > + > + return ret; > +} > + > +static int rbd_create(const char *filename, QEMUOptionParameter *options) > +{ > + int64_t bytes = 0; > + int64_t objsize; > + uint64_t size; > + time_t mtime; > + uint8_t obj_order = RBD_DEFAULT_OBJ_ORDER; > + char pool[RBD_MAX_SEG_NAME_SIZE]; > + char n[RBD_MAX_SEG_NAME_SIZE]; > + char name[RBD_MAX_SEG_NAME_SIZE]; > + RbdHeader1 header; > + rados_pool_t p; > + int name_len; > + int ret; > + > + if ((name_len = rbd_parsename(filename, pool, name)) < 0) { name_len is unused. > + return -EINVAL; > + } > + > + snprintf(n, RBD_MAX_SEG_NAME_SIZE, "%s%s", name, RBD_SUFFIX); n should probably be some bytes longer than name to contain RBD_SUFFIX additionally. Otherwise, checking the return value might be a good idea. > + > + /* Read out options */ > + while (options && options->name) { > + if (!strcmp(options->name, BLOCK_OPT_SIZE)) { > + bytes = options->value.n; > + } else if (!strcmp(options->name, BLOCK_OPT_CLUSTER_SIZE)) { > + if (options->value.n) { > + objsize = options->value.n; > + if (!objsize || ((objsize - 1) & objsize)) { /* not a power of 2? */ !objsize can't happen here, it's the if condition above. > + fprintf(stderr, "obj size needs to be power of 2\n"); > + return -EINVAL; > + } > + if (objsize < 4096) { > + fprintf(stderr, "obj size too small\n"); > + return -EINVAL; > + } > + > + for (obj_order = 0; obj_order < 64; obj_order++) { > + if (objsize == 1) > + break; Missing braces. > + objsize >>= 1; > + } > + } > + } > + options++; > + } > + > + memset(&header, 0, sizeof(header)); > + pstrcpy(header.text, sizeof(header.text), rbd_text); > + pstrcpy(header.signature, sizeof(header.signature), rbd_signature); > + pstrcpy(header.version, sizeof(header.version), rbd_version); > + header.image_size = bytes; > + cpu_to_le64s((uint64_t *) & header.image_size); > + header.options.order = obj_order; > + header.options.crypt_type = RBD_CRYPT_NONE; > + header.options.comp_type = RBD_COMP_NONE; > + header.snap_seq = 0; > + header.snap_count = 0; > + cpu_to_le32s(&header.snap_count); > + > + if (rados_initialize(0, NULL) < 0) { > + fprintf(stderr, "error initializing\n"); > + return -EIO; > + } > + > + if (rados_open_pool(pool, &p)) { > + fprintf(stderr, "error opening pool %s\n", pool); > + return -EIO; No need for rados_deinitialize() here? > + } > + > + /* check for existing rbd header file */ > + ret = rados_stat(p, n, &size, &mtime); > + if (ret == 0) { > + ret=-EEXIST; > + goto done; > + } > + > + /* create header file */ > + ret = rados_write(p, n, 0, (const char *)&header, sizeof(header)); > + if (ret < 0) { > + goto done; > + } > + > + ret = rbd_register_image(p, name); > +done: > + rados_close_pool(p); > + rados_deinitialize(); > + > + return ret; > +} > + > +static int rbd_open(BlockDriverState *bs, const char *filename, int flags) > +{ > + RBDRVRBDState *s = bs->opaque; > + char pool[RBD_MAX_SEG_NAME_SIZE]; > + char n[RBD_MAX_SEG_NAME_SIZE]; > + char hbuf[4096]; > + > + if ((s->name_len = rbd_parsename(filename, pool, s->name)) < 0) { > + return -EINVAL; > + } > + snprintf(n, RBD_MAX_SEG_NAME_SIZE, "%s%s", s->name, RBD_SUFFIX); > + > + if (rados_initialize(0, NULL) < 0) { > + fprintf(stderr, "error initializing\n"); > + return -EIO; > + } What error codes do these rados_* functions return? Can we return something more meaningful than -EIO? Same question for the following calls. > + > + if (rados_open_pool(pool, &s->pool)) { > + fprintf(stderr, "error opening pool %s\n", pool); > + return -EIO; > + } rados_deinitialize? (same for the following return statements) > + > + if (rados_read(s->pool, n, 0, hbuf, 4096) < 0) { > + fprintf(stderr, "error reading header from %s\n", s->name); > + return -EIO; > + } > + if (!strncmp(hbuf + 64, rbd_signature, 4)) { > + if (!strncmp(hbuf + 68, rbd_version, 8)) { > + RbdHeader1 *header; > + > + header = (RbdHeader1 *) hbuf; > + le64_to_cpus((uint64_t *) & header->image_size); > + s->size = header->image_size; > + s->objsize = 1 << header->options.order; > + } else { > + fprintf(stderr, "Unknown image version %s\n", hbuf + 68); > + return -EIO; > + } > + } else { > + fprintf(stderr, "Invalid header signature %s\n", hbuf + 64); > + return -EIO; > + } Can't you go on like above, i.e. check the string and if the check fails return an error? Would probably be more readable than this nesting. Also, EIO for a failed string comparison is surely not the right error code? > + > + return 0; > +} > + > +static void rbd_close(BlockDriverState *bs) > +{ > + RBDRVRBDState *s = bs->opaque; > + > + rados_close_pool(s->pool); > + rados_deinitialize(); > +} > + > +static int rbd_rw(BlockDriverState *bs, int64_t sector_num, > + uint8_t *buf, int nb_sectors, int write) > +{ > + RBDRVRBDState *s = bs->opaque; > + char n[RBD_MAX_SEG_NAME_SIZE]; > + > + int64_t segnr, segoffs, segsize, r; > + int64_t off, size; > + > + off = sector_num * 512; > + size = nb_sectors * 512; Please use BDRV_SECTOR_SIZE. > + segnr = (int64_t) (off / s->objsize); > + segoffs = (int64_t) (off % s->objsize); > + segsize = (int64_t) (s->objsize - segoffs); Why the type casts? Do they make any difference in this place? > + > + while (size > 0) { > + if (size < segsize) { > + segsize = size; > + } > + > + snprintf(n, RBD_MAX_SEG_NAME_SIZE, "%s.%012llx", s->name, > + (long long unsigned int)segnr); n is possibly to short, see above. Also you could use PRIx64 instead of casting to long long. > + > + if (write) { > + if ((r = rados_write(s->pool, n, segoffs, (const char *)buf, > + segsize)) < 0) { > + return r; > + } > + } else { > + r = rados_read(s->pool, n, segoffs, (char *)buf, segsize); > + if (r == -ENOENT) { > + memset(buf, 0, segsize); > + } else if (r < 0) { > + return(r); These brackets look strange (and inconsistent with all other returns in the code) > + } else if (r < segsize) { > + memset(buf + r, 0, segsize - r); > + } > + r = segsize; What is this good for? r isn't used before it's overwritten in the next loop iteration. > + } > + > + buf += segsize; > + size -= segsize; > + segoffs = 0; > + segsize = s->objsize; > + segnr++; > + } > + > + return (0); Again brackets. > +} > + > +static int rbd_read(BlockDriverState *bs, int64_t sector_num, > + uint8_t *buf, int nb_sectors) > +{ > + return rbd_rw(bs, sector_num, buf, nb_sectors, 0); > +} > + > +static int rbd_write(BlockDriverState *bs, int64_t sector_num, > + const uint8_t *buf, int nb_sectors) > +{ > + return rbd_rw(bs, sector_num, (uint8_t *) buf, nb_sectors, 1); > +} > + > +static void rbd_aio_cancel(BlockDriverAIOCB *blockacb) > +{ > + RBDAIOCB *acb = (RBDAIOCB *) blockacb; > + qemu_bh_delete(acb->bh); > + acb->bh = NULL; > + qemu_aio_release(acb); > +} > + > +static AIOPool rbd_aio_pool = { > + .aiocb_size = sizeof(RBDAIOCB), > + .cancel = rbd_aio_cancel, > +}; > + > +/* This is the callback function for rados_aio_read and _write */ > +static void rbd_finish_aiocb(rados_completion_t c, RADOSCB *rcb) > +{ > + RBDAIOCB *acb = rcb->acb; > + int64_t r; > + int i; > + > + acb->aiocnt--; > + r = rados_aio_get_return_value(c); > + rados_aio_release(c); > + if (acb->write) { > + if (r < 0) { > + acb->ret = r; > + acb->error = 1; > + } else if (!acb->error) { > + acb->ret += rcb->segsize; > + } > + } else { > + if (r == -ENOENT) { > + memset(rcb->buf, 0, rcb->segsize); > + if (!acb->error) { > + acb->ret += rcb->segsize; > + } > + } else if (r < 0) { > + acb->ret = r; > + acb->error = 1; > + } else if (r < rcb->segsize) { > + memset(rcb->buf + r, 0, rcb->segsize - r); > + if (!acb->error) { > + acb->ret += rcb->segsize; > + } > + } else if (!acb->error) { > + acb->ret += r; > + } > + } > + qemu_free(rcb); > + i = 0; > + if (!acb->aiocnt && acb->bh) { > + qemu_bh_schedule(acb->bh); > + } > +} > + > +/* Callback when all queued rados_aio requests are complete */ > +static void rbd_aio_bh_cb(void *opaque) > +{ > + RBDAIOCB *acb = opaque; > + > + if (!acb->write) { > + qemu_iovec_from_buffer(acb->qiov, acb->bounce, acb->qiov->size); > + } > + qemu_vfree(acb->bounce); > + acb->common.cb(acb->common.opaque, (acb->ret > 0 ? 0 : acb->ret)); > + qemu_bh_delete(acb->bh); > + acb->bh = NULL; > + qemu_aio_release(acb); > +} > + > +static BlockDriverAIOCB *rbd_aio_rw_vector(BlockDriverState *bs, > + int64_t sector_num, > + QEMUIOVector *qiov, > + int nb_sectors, > + BlockDriverCompletionFunc *cb, > + void *opaque, int write) For this function the same applies as for the synchronous one. > +{ > + RBDAIOCB *acb; > + RADOSCB *rcb; > + rados_completion_t c; > + char n[RBD_MAX_SEG_NAME_SIZE]; > + int64_t segnr, segoffs, segsize, last_segnr; > + int64_t off, size; > + char *buf; > + > + RBDRVRBDState *s = bs->opaque; > + > + acb = qemu_aio_get(&rbd_aio_pool, bs, cb, opaque); > + acb->write = write; > + acb->qiov = qiov; > + acb->bounce = qemu_blockalign(bs, qiov->size); > + acb->aiocnt = 0; > + acb->ret = 0; > + acb->error = 0; > + > + if (!acb->bh) { > + acb->bh = qemu_bh_new(rbd_aio_bh_cb, acb); > + } > + > + if (write) { > + qemu_iovec_to_buffer(acb->qiov, acb->bounce); > + } > + > + buf = acb->bounce; > + > + off = sector_num * 512; > + size = nb_sectors * 512; > + segnr = (int64_t) (off / s->objsize); > + segoffs = (int64_t) (off % s->objsize); > + segsize = (int64_t) (s->objsize - segoffs); > + > + last_segnr = ((off + size - 1) / s->objsize); > + acb->aiocnt = (last_segnr - segnr) + 1; > + > + while (size > 0) { > + if (size < segsize) { > + segsize = size; > + } > + > + snprintf(n, RBD_MAX_SEG_NAME_SIZE, "%s.%012llx", s->name, > + (long long unsigned int)segnr); > + > + rcb = qemu_malloc(sizeof(RADOSCB)); > + rcb->done = 0; > + rcb->acb = acb; > + rcb->segsize = segsize; > + rcb->buf = buf; > + > + if (write) { > + rados_aio_create_completion(rcb, NULL, > + (rados_callback_t) rbd_finish_aiocb, &c); > + rados_aio_write(s->pool, n, segoffs, buf, segsize, c); > + } else { > + rados_aio_create_completion(rcb, (rados_callback_t) rbd_finish_aiocb, > + NULL, &c); > + rados_aio_read(s->pool, n, segoffs, buf, segsize, c); > + } > + > + buf += segsize; > + size -= segsize; > + segoffs = 0; > + segsize = s->objsize; > + segnr++; > + } > + > + return &acb->common; > +} > + > +static BlockDriverAIOCB *rbd_aio_readv(BlockDriverState * bs, > + int64_t sector_num, QEMUIOVector * qiov, > + int nb_sectors, > + BlockDriverCompletionFunc * cb, > + void *opaque) > +{ > + return rbd_aio_rw_vector(bs, sector_num, qiov, nb_sectors, cb, opaque, 0); > +} > + > +static BlockDriverAIOCB *rbd_aio_writev(BlockDriverState * bs, > + int64_t sector_num, QEMUIOVector * qiov, > + int nb_sectors, > + BlockDriverCompletionFunc * cb, > + void *opaque) > +{ > + return rbd_aio_rw_vector(bs, sector_num, qiov, nb_sectors, cb, opaque, 1); > +} > + > +static int rbd_getinfo(BlockDriverState * bs, BlockDriverInfo * bdi) > +{ > + RBDRVRBDState *s = bs->opaque; > + bdi->cluster_size = s->objsize; > + return 0; > +} > + > +static int64_t rbd_getlength(BlockDriverState * bs) > +{ > + RBDRVRBDState *s = bs->opaque; > + > + return s->size; > +} > + > +static QEMUOptionParameter rbd_create_options[] = { > + { > + .name = BLOCK_OPT_SIZE, > + .type = OPT_SIZE, > + .help = "Virtual disk size" > + }, > + { > + .name = BLOCK_OPT_CLUSTER_SIZE, > + .type = OPT_SIZE, > + .help = "RBD object size" > + }, > + {NULL} > +}; > + > +static BlockDriver bdrv_rbd = { > + .format_name = "rbd", > + .instance_size = sizeof(RBDRVRBDState), > + .bdrv_file_open = rbd_open, > + .bdrv_read = rbd_read, > + .bdrv_write = rbd_write, > + .bdrv_close = rbd_close, > + .bdrv_create = rbd_create, > + .bdrv_get_info = rbd_getinfo, > + .create_options = rbd_create_options, > + .bdrv_getlength = rbd_getlength, > + .protocol_name = "rbd", > + > + .bdrv_aio_readv = rbd_aio_readv, > + .bdrv_aio_writev = rbd_aio_writev, > +}; Could you align the = to be on the same column? > + > +static void bdrv_rbd_init(void) > +{ > + bdrv_register(&bdrv_rbd); > +} > + > +block_init(bdrv_rbd_init); > diff --git a/block/rbd_types.h b/block/rbd_types.h > new file mode 100644 > index 0000000..3a16a26 > --- /dev/null > +++ b/block/rbd_types.h > @@ -0,0 +1,52 @@ We need a header comment with copyright/license. > +#ifndef QEMU_BLOCK_RBD_TYPES_H > +#define QEMU_BLOCK_RBD_TYPES_H > + > + > +/* > + * rbd image 'foo' consists of objects > + * foo.rbd - image metadata > + * foo.00000000 Trailing whitespace. > + * foo.00000001 > + * ... - data > + */ > + > +#define RBD_SUFFIX ".rbd" > +#define RBD_DIRECTORY "rbd_directory" > + > +#define RBD_DEFAULT_OBJ_ORDER 22 /* 4MB */ > + > +#define RBD_MAX_OBJ_NAME_SIZE 96 > +#define RBD_MAX_SEG_NAME_SIZE 128 > + > +#define RBD_COMP_NONE 0 > +#define RBD_CRYPT_NONE 0 > + > +static const char rbd_text[] = "<<< Rados Block Device Image >>>\n"; > +static const char rbd_signature[] = "RBD"; > +static const char rbd_version[] = "001.004"; Move these to rbd.c? > + > +struct rbd_obj_snap_ondisk { > + uint64_t id; > + uint64_t image_size; > +} __attribute__((packed)); > + > +struct rbd_obj_header_ondisk { > + char text[64]; > + char signature[4]; > + char version[8]; > + struct { > + uint8_t order; > + uint8_t crypt_type; > + uint8_t comp_type; > + uint8_t unused; > + } __attribute__((packed)) options; > + uint64_t image_size; > + uint64_t snap_seq; > + uint32_t snap_count; > + uint32_t reserved; > + uint64_t snap_names_len; > + struct rbd_obj_snap_ondisk snaps[0]; > +} __attribute__((packed)); > + > + > +#endif > diff --git a/configure b/configure > index 3cd2c5f..7d70bf8 100755 > --- a/configure > +++ b/configure > @@ -299,6 +299,7 @@ pkgversion="" > check_utests="no" > user_pie="no" > zero_malloc="" > +rbd="no" Shouldn't it default to auto-detection? (rbd="") > > # OS specific > if check_define __linux__ ; then > @@ -660,6 +661,8 @@ for opt do > ;; > --enable-vhost-net) vhost_net="yes" > ;; > + --enable-rbd) rbd="yes" > + ;; Would need a --disable-rbd flag then, too. > *) echo "ERROR: unknown option $opt"; show_help="yes" > ;; > esac > @@ -826,6 +829,7 @@ echo " --enable-docs enable documentation build" > echo " --disable-docs disable documentation build" > echo " --disable-vhost-net disable vhost-net acceleration support" > echo " --enable-vhost-net enable vhost-net acceleration support" > +echo " --enable-rbd enable building the rados block device (rbd)" > echo "" > echo "NOTE: The object files are built at the place where configure is launched" > exit 1 > @@ -1579,6 +1583,25 @@ if test "$mingw32" != yes -a "$pthread" = no; then > fi > > ########################################## > +# rbd probe > +if test "$rbd" != "no" ; then > + cat > $TMPC <<EOF > +#include <stdio.h> > +#include <rados/librados.h> > +int main(void) { rados_initialize(0, NULL); return 0; } > +EOF > + if compile_prog "" "-lrados -lcrypto" ; then > + rbd=yes > + LIBS="$LIBS -lrados -lcrypto" Other block features like curl or Linux AIO use $lib_tools and $lib_softmmu here. > + else > + if test "$rbd" = "yes" ; then > + feature_not_found "rados block device" > + fi > + rbd=no > + fi > +fi But if you didn't want to auto-detect, this is more complicated than it needs to be. Kevin
Hi Kevin, thanks for your review notes. Yehuda and I have already worked this into the git tree on the ceph site. I'll do some testing on Monday. After that I'll send an updated patch. Regards, Christian 2010/5/28 Kevin Wolf <kwolf@redhat.com>: > Am 27.05.2010 21:11, schrieb Christian Brunner: >> This is a block driver for the distributed file system Ceph >> (http://ceph.newdream.net/). This driver uses librados (which >> is part of the Ceph server) for direct access to the Ceph object >> store and is running entirely in userspace. Therefore it is >> called "rbd" - rados block device. >> >> To compile the driver a recent version of ceph (unstable/testin git >> head or 0.20.3 once it is released) is needed and you have to >> "--enable-rbd" when running configure. >> >> Additional information is available on the Ceph-Wiki: >> >> http://ceph.newdream.net/wiki/Kvm-rbd >> >> The patch is based on git://repo.or.cz/qemu/kevin.git block > > Signed-off-by line is missing. > >> --- >> Makefile | 3 + >> Makefile.objs | 1 + >> block/rbd.c | 584 +++++++++++++++++++++++++++++++++++++++++++++++++++++ >> block/rbd_types.h | 52 +++++ >> configure | 27 +++ >> 5 files changed, 667 insertions(+), 0 deletions(-) >> create mode 100644 block/rbd.c >> create mode 100644 block/rbd_types.h >> >> diff --git a/Makefile b/Makefile >> index 7986bf6..8d09612 100644 >> --- a/Makefile >> +++ b/Makefile >> @@ -27,6 +27,9 @@ configure: ; >> $(call set-vpath, $(SRC_PATH):$(SRC_PATH)/hw) >> >> LIBS+=-lz $(LIBS_TOOLS) >> +ifdef CONFIG_RBD >> +LIBS+=-lrados >> +endif > > You already write the -lrados option to config-host.mak in configure, so > this looks unnecessary. > >> >> ifdef BUILD_DOCS >> DOCS=qemu-doc.html qemu-tech.html qemu.1 qemu-img.1 qemu-nbd.8 >> diff --git a/Makefile.objs b/Makefile.objs >> index 1a942e5..08dc11f 100644 >> --- a/Makefile.objs >> +++ b/Makefile.objs >> @@ -18,6 +18,7 @@ block-nested-y += parallels.o nbd.o blkdebug.o >> block-nested-$(CONFIG_WIN32) += raw-win32.o >> block-nested-$(CONFIG_POSIX) += raw-posix.o >> block-nested-$(CONFIG_CURL) += curl.o >> +block-nested-$(CONFIG_RBD) += rbd.o >> >> block-obj-y += $(addprefix block/, $(block-nested-y)) >> >> diff --git a/block/rbd.c b/block/rbd.c >> new file mode 100644 >> index 0000000..375ae9d >> --- /dev/null >> +++ b/block/rbd.c >> @@ -0,0 +1,584 @@ >> +/* >> + * QEMU Block driver for RADOS (Ceph) >> + * >> + * Copyright (C) 2010 Christian Brunner <chb@muc.de> >> + * >> + * This work is licensed under the terms of the GNU GPL, version 2. See >> + * the COPYING file in the top-level directory. >> + * >> + */ >> + >> +#include "qemu-common.h" >> +#include <sys/types.h> >> +#include <stdbool.h> >> + >> +#include <qemu-common.h> >> + >> +#include "rbd_types.h" >> +#include "module.h" >> +#include "block_int.h" >> + >> +#include <stdio.h> >> +#include <stdlib.h> >> +#include <rados/librados.h> >> + >> +#include <signal.h> >> + >> +/* >> + * When specifying the image filename use: >> + * >> + * rbd:poolname/devicename >> + * >> + * poolname must be the name of an existing rados pool >> + * >> + * devicename is the basename for all objects used to >> + * emulate the raw device. >> + * >> + * Metadata information (image size, ...) is stored in an >> + * object with the name "devicename.rbd". >> + * >> + * The raw device is split into 4MB sized objects by default. >> + * The sequencenumber is encoded in a 12 byte long hex-string, >> + * and is attached to the devicename, separated by a dot. >> + * e.g. "devicename.1234567890ab" >> + * >> + */ >> + >> +#define OBJ_MAX_SIZE (1UL << OBJ_DEFAULT_OBJ_ORDER) >> + >> +typedef struct RBDAIOCB { >> + BlockDriverAIOCB common; >> + QEMUBH *bh; >> + int ret; >> + QEMUIOVector *qiov; >> + char *bounce; >> + int write; >> + int64_t sector_num; >> + int aiocnt; >> + int error; >> +} RBDAIOCB; >> + >> +typedef struct RADOSCB { >> + int rcbid; >> + RBDAIOCB *acb; >> + int done; >> + int64_t segsize; >> + char *buf; >> +} RADOSCB; >> + >> +typedef struct RBDRVRBDState { >> + rados_pool_t pool; >> + char name[RBD_MAX_OBJ_NAME_SIZE]; >> + int name_len; > > name_len looks unused. > >> + uint64_t size; >> + uint64_t objsize; >> +} RBDRVRBDState; > > Hm, you mean BDRVRBDState? > > Maybe ceph would have been a better driver name to avoid such type > names. ;-) > >> + >> +typedef struct rbd_obj_header_ondisk RbdHeader1; >> + >> +static int rbd_parsename(const char *filename, char *pool, char *name) >> +{ >> + const char *rbdname; >> + char *p, *n; >> + int l; >> + >> + if (!strstart(filename, "rbd:", &rbdname)) { >> + return -EINVAL; >> + } >> + >> + pstrcpy(pool, 2 * RBD_MAX_SEG_NAME_SIZE, rbdname); > > Why twice the size? The callers pass a char[RBD_MAX_SEG_NAME_SIZE], so > doesn't this allow buffer overflows? > >> + p = strchr(pool, '/'); >> + if (p == NULL) { >> + return -EINVAL; >> + } >> + >> + *p = '\0'; >> + n = ++p; > > Why introduce a new variable here? p isn't used any more afterwards. > >> + >> + l = strlen(n); >> + >> + if (l > RBD_MAX_OBJ_NAME_SIZE) { >> + fprintf(stderr, "object name to long\n"); > > Off by one, you need to consider the trailing '\0'. > > Also, please use error_report instead of fprintf(stderr, ...) for real > error messages. Directly printing to stderr is okay for debug code. > >> + return -EINVAL; >> + } else if (l <= 0) { >> + fprintf(stderr, "object name to short\n"); >> + return -EINVAL; >> + } >> + >> + strcpy(name, n); >> + >> + return l; >> +} >> + >> +static int create_tmap_op(uint8_t op, const char *name, char **tmap_desc) >> +{ >> + uint32_t len = strlen(name); >> + uint32_t total_len = 1 + (sizeof(uint32_t) + len) + sizeof(uint32_t); /* encoding op + name + empty buffer */ > > This is more than 80 characters. > >> + char *desc; >> + >> + desc = qemu_malloc(total_len); >> + if (!desc) { >> + return -ENOMEM; >> + } > > qemu_malloc never returns NULL. > >> + >> + *tmap_desc = desc; >> + >> + *desc = op; >> + desc++; >> + memcpy(desc, &len, sizeof(len)); >> + desc += sizeof(len); >> + memcpy(desc, name, len); >> + desc += len; >> + len = 0; >> + memcpy(desc, &len, sizeof(len)); >> + desc += sizeof(len); >> + >> + return desc - *tmap_desc; >> +} >> + >> +static void free_tmap_op(char *tmap_desc) >> +{ >> + qemu_free(tmap_desc); >> +} >> + >> +static int rbd_register_image(rados_pool_t pool, const char *name) >> +{ >> + char *tmap_desc; >> + const char *dir = RBD_DIRECTORY; >> + int ret; >> + >> + ret = create_tmap_op(CEPH_OSD_TMAP_SET, name, &tmap_desc); >> + if (ret < 0) { >> + return ret; >> + } >> + >> + ret = rados_tmap_update(pool, dir, tmap_desc, ret); >> + free_tmap_op(tmap_desc); >> + >> + return ret; >> +} >> + >> +static int rbd_create(const char *filename, QEMUOptionParameter *options) >> +{ >> + int64_t bytes = 0; >> + int64_t objsize; >> + uint64_t size; >> + time_t mtime; >> + uint8_t obj_order = RBD_DEFAULT_OBJ_ORDER; >> + char pool[RBD_MAX_SEG_NAME_SIZE]; >> + char n[RBD_MAX_SEG_NAME_SIZE]; >> + char name[RBD_MAX_SEG_NAME_SIZE]; >> + RbdHeader1 header; >> + rados_pool_t p; >> + int name_len; >> + int ret; >> + >> + if ((name_len = rbd_parsename(filename, pool, name)) < 0) { > > name_len is unused. > >> + return -EINVAL; >> + } >> + >> + snprintf(n, RBD_MAX_SEG_NAME_SIZE, "%s%s", name, RBD_SUFFIX); > > n should probably be some bytes longer than name to contain RBD_SUFFIX > additionally. Otherwise, checking the return value might be a good idea. > >> + >> + /* Read out options */ >> + while (options && options->name) { >> + if (!strcmp(options->name, BLOCK_OPT_SIZE)) { >> + bytes = options->value.n; >> + } else if (!strcmp(options->name, BLOCK_OPT_CLUSTER_SIZE)) { >> + if (options->value.n) { >> + objsize = options->value.n; >> + if (!objsize || ((objsize - 1) & objsize)) { /* not a power of 2? */ > > !objsize can't happen here, it's the if condition above. > >> + fprintf(stderr, "obj size needs to be power of 2\n"); >> + return -EINVAL; >> + } >> + if (objsize < 4096) { >> + fprintf(stderr, "obj size too small\n"); >> + return -EINVAL; >> + } >> + >> + for (obj_order = 0; obj_order < 64; obj_order++) { >> + if (objsize == 1) >> + break; > > Missing braces. > >> + objsize >>= 1; >> + } >> + } >> + } >> + options++; >> + } >> + >> + memset(&header, 0, sizeof(header)); >> + pstrcpy(header.text, sizeof(header.text), rbd_text); >> + pstrcpy(header.signature, sizeof(header.signature), rbd_signature); >> + pstrcpy(header.version, sizeof(header.version), rbd_version); >> + header.image_size = bytes; >> + cpu_to_le64s((uint64_t *) & header.image_size); >> + header.options.order = obj_order; >> + header.options.crypt_type = RBD_CRYPT_NONE; >> + header.options.comp_type = RBD_COMP_NONE; >> + header.snap_seq = 0; >> + header.snap_count = 0; >> + cpu_to_le32s(&header.snap_count); >> + >> + if (rados_initialize(0, NULL) < 0) { >> + fprintf(stderr, "error initializing\n"); >> + return -EIO; >> + } >> + >> + if (rados_open_pool(pool, &p)) { >> + fprintf(stderr, "error opening pool %s\n", pool); >> + return -EIO; > > No need for rados_deinitialize() here? > >> + } >> + >> + /* check for existing rbd header file */ >> + ret = rados_stat(p, n, &size, &mtime); >> + if (ret == 0) { >> + ret=-EEXIST; >> + goto done; >> + } >> + >> + /* create header file */ >> + ret = rados_write(p, n, 0, (const char *)&header, sizeof(header)); >> + if (ret < 0) { >> + goto done; >> + } >> + >> + ret = rbd_register_image(p, name); >> +done: >> + rados_close_pool(p); >> + rados_deinitialize(); >> + >> + return ret; >> +} >> + >> +static int rbd_open(BlockDriverState *bs, const char *filename, int flags) >> +{ >> + RBDRVRBDState *s = bs->opaque; >> + char pool[RBD_MAX_SEG_NAME_SIZE]; >> + char n[RBD_MAX_SEG_NAME_SIZE]; >> + char hbuf[4096]; >> + >> + if ((s->name_len = rbd_parsename(filename, pool, s->name)) < 0) { >> + return -EINVAL; >> + } >> + snprintf(n, RBD_MAX_SEG_NAME_SIZE, "%s%s", s->name, RBD_SUFFIX); >> + >> + if (rados_initialize(0, NULL) < 0) { >> + fprintf(stderr, "error initializing\n"); >> + return -EIO; >> + } > > What error codes do these rados_* functions return? Can we return > something more meaningful than -EIO? Same question for the following calls. > >> + >> + if (rados_open_pool(pool, &s->pool)) { >> + fprintf(stderr, "error opening pool %s\n", pool); >> + return -EIO; >> + } > > rados_deinitialize? (same for the following return statements) > >> + >> + if (rados_read(s->pool, n, 0, hbuf, 4096) < 0) { >> + fprintf(stderr, "error reading header from %s\n", s->name); >> + return -EIO; >> + } >> + if (!strncmp(hbuf + 64, rbd_signature, 4)) { >> + if (!strncmp(hbuf + 68, rbd_version, 8)) { >> + RbdHeader1 *header; >> + >> + header = (RbdHeader1 *) hbuf; >> + le64_to_cpus((uint64_t *) & header->image_size); >> + s->size = header->image_size; >> + s->objsize = 1 << header->options.order; >> + } else { >> + fprintf(stderr, "Unknown image version %s\n", hbuf + 68); >> + return -EIO; >> + } >> + } else { >> + fprintf(stderr, "Invalid header signature %s\n", hbuf + 64); >> + return -EIO; >> + } > > Can't you go on like above, i.e. check the string and if the check fails > return an error? Would probably be more readable than this nesting. > > Also, EIO for a failed string comparison is surely not the right error code? > >> + >> + return 0; >> +} >> + >> +static void rbd_close(BlockDriverState *bs) >> +{ >> + RBDRVRBDState *s = bs->opaque; >> + >> + rados_close_pool(s->pool); >> + rados_deinitialize(); >> +} >> + >> +static int rbd_rw(BlockDriverState *bs, int64_t sector_num, >> + uint8_t *buf, int nb_sectors, int write) >> +{ >> + RBDRVRBDState *s = bs->opaque; >> + char n[RBD_MAX_SEG_NAME_SIZE]; >> + >> + int64_t segnr, segoffs, segsize, r; >> + int64_t off, size; >> + >> + off = sector_num * 512; >> + size = nb_sectors * 512; > > Please use BDRV_SECTOR_SIZE. > >> + segnr = (int64_t) (off / s->objsize); >> + segoffs = (int64_t) (off % s->objsize); >> + segsize = (int64_t) (s->objsize - segoffs); > > Why the type casts? Do they make any difference in this place? > >> + >> + while (size > 0) { >> + if (size < segsize) { >> + segsize = size; >> + } >> + >> + snprintf(n, RBD_MAX_SEG_NAME_SIZE, "%s.%012llx", s->name, >> + (long long unsigned int)segnr); > > n is possibly to short, see above. Also you could use PRIx64 instead of > casting to long long. > >> + >> + if (write) { >> + if ((r = rados_write(s->pool, n, segoffs, (const char *)buf, >> + segsize)) < 0) { >> + return r; >> + } >> + } else { >> + r = rados_read(s->pool, n, segoffs, (char *)buf, segsize); >> + if (r == -ENOENT) { >> + memset(buf, 0, segsize); >> + } else if (r < 0) { >> + return(r); > > These brackets look strange (and inconsistent with all other returns in > the code) > >> + } else if (r < segsize) { >> + memset(buf + r, 0, segsize - r); >> + } >> + r = segsize; > > What is this good for? r isn't used before it's overwritten in the next > loop iteration. > >> + } >> + >> + buf += segsize; >> + size -= segsize; >> + segoffs = 0; >> + segsize = s->objsize; >> + segnr++; >> + } >> + >> + return (0); > > Again brackets. > >> +} >> + >> +static int rbd_read(BlockDriverState *bs, int64_t sector_num, >> + uint8_t *buf, int nb_sectors) >> +{ >> + return rbd_rw(bs, sector_num, buf, nb_sectors, 0); >> +} >> + >> +static int rbd_write(BlockDriverState *bs, int64_t sector_num, >> + const uint8_t *buf, int nb_sectors) >> +{ >> + return rbd_rw(bs, sector_num, (uint8_t *) buf, nb_sectors, 1); >> +} >> + >> +static void rbd_aio_cancel(BlockDriverAIOCB *blockacb) >> +{ >> + RBDAIOCB *acb = (RBDAIOCB *) blockacb; >> + qemu_bh_delete(acb->bh); >> + acb->bh = NULL; >> + qemu_aio_release(acb); >> +} >> + >> +static AIOPool rbd_aio_pool = { >> + .aiocb_size = sizeof(RBDAIOCB), >> + .cancel = rbd_aio_cancel, >> +}; >> + >> +/* This is the callback function for rados_aio_read and _write */ >> +static void rbd_finish_aiocb(rados_completion_t c, RADOSCB *rcb) >> +{ >> + RBDAIOCB *acb = rcb->acb; >> + int64_t r; >> + int i; >> + >> + acb->aiocnt--; >> + r = rados_aio_get_return_value(c); >> + rados_aio_release(c); >> + if (acb->write) { >> + if (r < 0) { >> + acb->ret = r; >> + acb->error = 1; >> + } else if (!acb->error) { >> + acb->ret += rcb->segsize; >> + } >> + } else { >> + if (r == -ENOENT) { >> + memset(rcb->buf, 0, rcb->segsize); >> + if (!acb->error) { >> + acb->ret += rcb->segsize; >> + } >> + } else if (r < 0) { >> + acb->ret = r; >> + acb->error = 1; >> + } else if (r < rcb->segsize) { >> + memset(rcb->buf + r, 0, rcb->segsize - r); >> + if (!acb->error) { >> + acb->ret += rcb->segsize; >> + } >> + } else if (!acb->error) { >> + acb->ret += r; >> + } >> + } >> + qemu_free(rcb); >> + i = 0; >> + if (!acb->aiocnt && acb->bh) { >> + qemu_bh_schedule(acb->bh); >> + } >> +} >> + >> +/* Callback when all queued rados_aio requests are complete */ >> +static void rbd_aio_bh_cb(void *opaque) >> +{ >> + RBDAIOCB *acb = opaque; >> + >> + if (!acb->write) { >> + qemu_iovec_from_buffer(acb->qiov, acb->bounce, acb->qiov->size); >> + } >> + qemu_vfree(acb->bounce); >> + acb->common.cb(acb->common.opaque, (acb->ret > 0 ? 0 : acb->ret)); >> + qemu_bh_delete(acb->bh); >> + acb->bh = NULL; >> + qemu_aio_release(acb); >> +} >> + >> +static BlockDriverAIOCB *rbd_aio_rw_vector(BlockDriverState *bs, >> + int64_t sector_num, >> + QEMUIOVector *qiov, >> + int nb_sectors, >> + BlockDriverCompletionFunc *cb, >> + void *opaque, int write) > > For this function the same applies as for the synchronous one. > >> +{ >> + RBDAIOCB *acb; >> + RADOSCB *rcb; >> + rados_completion_t c; >> + char n[RBD_MAX_SEG_NAME_SIZE]; >> + int64_t segnr, segoffs, segsize, last_segnr; >> + int64_t off, size; >> + char *buf; >> + >> + RBDRVRBDState *s = bs->opaque; >> + >> + acb = qemu_aio_get(&rbd_aio_pool, bs, cb, opaque); >> + acb->write = write; >> + acb->qiov = qiov; >> + acb->bounce = qemu_blockalign(bs, qiov->size); >> + acb->aiocnt = 0; >> + acb->ret = 0; >> + acb->error = 0; >> + >> + if (!acb->bh) { >> + acb->bh = qemu_bh_new(rbd_aio_bh_cb, acb); >> + } >> + >> + if (write) { >> + qemu_iovec_to_buffer(acb->qiov, acb->bounce); >> + } >> + >> + buf = acb->bounce; >> + >> + off = sector_num * 512; >> + size = nb_sectors * 512; >> + segnr = (int64_t) (off / s->objsize); >> + segoffs = (int64_t) (off % s->objsize); >> + segsize = (int64_t) (s->objsize - segoffs); >> + >> + last_segnr = ((off + size - 1) / s->objsize); >> + acb->aiocnt = (last_segnr - segnr) + 1; >> + >> + while (size > 0) { >> + if (size < segsize) { >> + segsize = size; >> + } >> + >> + snprintf(n, RBD_MAX_SEG_NAME_SIZE, "%s.%012llx", s->name, >> + (long long unsigned int)segnr); >> + >> + rcb = qemu_malloc(sizeof(RADOSCB)); >> + rcb->done = 0; >> + rcb->acb = acb; >> + rcb->segsize = segsize; >> + rcb->buf = buf; >> + >> + if (write) { >> + rados_aio_create_completion(rcb, NULL, >> + (rados_callback_t) rbd_finish_aiocb, &c); >> + rados_aio_write(s->pool, n, segoffs, buf, segsize, c); >> + } else { >> + rados_aio_create_completion(rcb, (rados_callback_t) rbd_finish_aiocb, >> + NULL, &c); >> + rados_aio_read(s->pool, n, segoffs, buf, segsize, c); >> + } >> + >> + buf += segsize; >> + size -= segsize; >> + segoffs = 0; >> + segsize = s->objsize; >> + segnr++; >> + } >> + >> + return &acb->common; >> +} >> + >> +static BlockDriverAIOCB *rbd_aio_readv(BlockDriverState * bs, >> + int64_t sector_num, QEMUIOVector * qiov, >> + int nb_sectors, >> + BlockDriverCompletionFunc * cb, >> + void *opaque) >> +{ >> + return rbd_aio_rw_vector(bs, sector_num, qiov, nb_sectors, cb, opaque, 0); >> +} >> + >> +static BlockDriverAIOCB *rbd_aio_writev(BlockDriverState * bs, >> + int64_t sector_num, QEMUIOVector * qiov, >> + int nb_sectors, >> + BlockDriverCompletionFunc * cb, >> + void *opaque) >> +{ >> + return rbd_aio_rw_vector(bs, sector_num, qiov, nb_sectors, cb, opaque, 1); >> +} >> + >> +static int rbd_getinfo(BlockDriverState * bs, BlockDriverInfo * bdi) >> +{ >> + RBDRVRBDState *s = bs->opaque; >> + bdi->cluster_size = s->objsize; >> + return 0; >> +} >> + >> +static int64_t rbd_getlength(BlockDriverState * bs) >> +{ >> + RBDRVRBDState *s = bs->opaque; >> + >> + return s->size; >> +} >> + >> +static QEMUOptionParameter rbd_create_options[] = { >> + { >> + .name = BLOCK_OPT_SIZE, >> + .type = OPT_SIZE, >> + .help = "Virtual disk size" >> + }, >> + { >> + .name = BLOCK_OPT_CLUSTER_SIZE, >> + .type = OPT_SIZE, >> + .help = "RBD object size" >> + }, >> + {NULL} >> +}; >> + >> +static BlockDriver bdrv_rbd = { >> + .format_name = "rbd", >> + .instance_size = sizeof(RBDRVRBDState), >> + .bdrv_file_open = rbd_open, >> + .bdrv_read = rbd_read, >> + .bdrv_write = rbd_write, >> + .bdrv_close = rbd_close, >> + .bdrv_create = rbd_create, >> + .bdrv_get_info = rbd_getinfo, >> + .create_options = rbd_create_options, >> + .bdrv_getlength = rbd_getlength, >> + .protocol_name = "rbd", >> + >> + .bdrv_aio_readv = rbd_aio_readv, >> + .bdrv_aio_writev = rbd_aio_writev, >> +}; > > Could you align the = to be on the same column? > >> + >> +static void bdrv_rbd_init(void) >> +{ >> + bdrv_register(&bdrv_rbd); >> +} >> + >> +block_init(bdrv_rbd_init); >> diff --git a/block/rbd_types.h b/block/rbd_types.h >> new file mode 100644 >> index 0000000..3a16a26 >> --- /dev/null >> +++ b/block/rbd_types.h >> @@ -0,0 +1,52 @@ > > We need a header comment with copyright/license. > >> +#ifndef QEMU_BLOCK_RBD_TYPES_H >> +#define QEMU_BLOCK_RBD_TYPES_H >> + >> + >> +/* >> + * rbd image 'foo' consists of objects >> + * foo.rbd - image metadata >> + * foo.00000000 > > Trailing whitespace. > >> + * foo.00000001 >> + * ... - data >> + */ >> + >> +#define RBD_SUFFIX ".rbd" >> +#define RBD_DIRECTORY "rbd_directory" >> + >> +#define RBD_DEFAULT_OBJ_ORDER 22 /* 4MB */ >> + >> +#define RBD_MAX_OBJ_NAME_SIZE 96 >> +#define RBD_MAX_SEG_NAME_SIZE 128 >> + >> +#define RBD_COMP_NONE 0 >> +#define RBD_CRYPT_NONE 0 >> + >> +static const char rbd_text[] = "<<< Rados Block Device Image >>>\n"; >> +static const char rbd_signature[] = "RBD"; >> +static const char rbd_version[] = "001.004"; > > Move these to rbd.c? > >> + >> +struct rbd_obj_snap_ondisk { >> + uint64_t id; >> + uint64_t image_size; >> +} __attribute__((packed)); >> + >> +struct rbd_obj_header_ondisk { >> + char text[64]; >> + char signature[4]; >> + char version[8]; >> + struct { >> + uint8_t order; >> + uint8_t crypt_type; >> + uint8_t comp_type; >> + uint8_t unused; >> + } __attribute__((packed)) options; >> + uint64_t image_size; >> + uint64_t snap_seq; >> + uint32_t snap_count; >> + uint32_t reserved; >> + uint64_t snap_names_len; >> + struct rbd_obj_snap_ondisk snaps[0]; >> +} __attribute__((packed)); >> + >> + >> +#endif >> diff --git a/configure b/configure >> index 3cd2c5f..7d70bf8 100755 >> --- a/configure >> +++ b/configure >> @@ -299,6 +299,7 @@ pkgversion="" >> check_utests="no" >> user_pie="no" >> zero_malloc="" >> +rbd="no" > > Shouldn't it default to auto-detection? (rbd="") > >> >> # OS specific >> if check_define __linux__ ; then >> @@ -660,6 +661,8 @@ for opt do >> ;; >> --enable-vhost-net) vhost_net="yes" >> ;; >> + --enable-rbd) rbd="yes" >> + ;; > > Would need a --disable-rbd flag then, too. > >> *) echo "ERROR: unknown option $opt"; show_help="yes" >> ;; >> esac >> @@ -826,6 +829,7 @@ echo " --enable-docs enable documentation build" >> echo " --disable-docs disable documentation build" >> echo " --disable-vhost-net disable vhost-net acceleration support" >> echo " --enable-vhost-net enable vhost-net acceleration support" >> +echo " --enable-rbd enable building the rados block device (rbd)" >> echo "" >> echo "NOTE: The object files are built at the place where configure is launched" >> exit 1 >> @@ -1579,6 +1583,25 @@ if test "$mingw32" != yes -a "$pthread" = no; then >> fi >> >> ########################################## >> +# rbd probe >> +if test "$rbd" != "no" ; then >> + cat > $TMPC <<EOF >> +#include <stdio.h> >> +#include <rados/librados.h> >> +int main(void) { rados_initialize(0, NULL); return 0; } >> +EOF >> + if compile_prog "" "-lrados -lcrypto" ; then >> + rbd=yes >> + LIBS="$LIBS -lrados -lcrypto" > > Other block features like curl or Linux AIO use $lib_tools and > $lib_softmmu here. > >> + else >> + if test "$rbd" = "yes" ; then >> + feature_not_found "rados block device" >> + fi >> + rbd=no >> + fi >> +fi > > But if you didn't want to auto-detect, this is more complicated than it > needs to be. > > Kevin > -- > To unsubscribe from this list: send the line "unsubscribe kvm" in > the body of a message to majordomo@vger.kernel.org > More majordomo info at http://vger.kernel.org/majordomo-info.html >
diff --git a/Makefile b/Makefile index 7986bf6..8d09612 100644 --- a/Makefile +++ b/Makefile @@ -27,6 +27,9 @@ configure: ; $(call set-vpath, $(SRC_PATH):$(SRC_PATH)/hw) LIBS+=-lz $(LIBS_TOOLS) +ifdef CONFIG_RBD +LIBS+=-lrados +endif ifdef BUILD_DOCS DOCS=qemu-doc.html qemu-tech.html qemu.1 qemu-img.1 qemu-nbd.8 diff --git a/Makefile.objs b/Makefile.objs index 1a942e5..08dc11f 100644 --- a/Makefile.objs +++ b/Makefile.objs @@ -18,6 +18,7 @@ block-nested-y += parallels.o nbd.o blkdebug.o block-nested-$(CONFIG_WIN32) += raw-win32.o block-nested-$(CONFIG_POSIX) += raw-posix.o block-nested-$(CONFIG_CURL) += curl.o +block-nested-$(CONFIG_RBD) += rbd.o block-obj-y += $(addprefix block/, $(block-nested-y)) diff --git a/block/rbd.c b/block/rbd.c new file mode 100644 index 0000000..375ae9d --- /dev/null +++ b/block/rbd.c @@ -0,0 +1,584 @@ +/* + * QEMU Block driver for RADOS (Ceph) + * + * Copyright (C) 2010 Christian Brunner <chb@muc.de> + * + * This work is licensed under the terms of the GNU GPL, version 2. See + * the COPYING file in the top-level directory. + * + */ + +#include "qemu-common.h" +#include <sys/types.h> +#include <stdbool.h> + +#include <qemu-common.h> + +#include "rbd_types.h" +#include "module.h" +#include "block_int.h" + +#include <stdio.h> +#include <stdlib.h> +#include <rados/librados.h> + +#include <signal.h> + +/* + * When specifying the image filename use: + * + * rbd:poolname/devicename + * + * poolname must be the name of an existing rados pool + * + * devicename is the basename for all objects used to + * emulate the raw device. + * + * Metadata information (image size, ...) is stored in an + * object with the name "devicename.rbd". + * + * The raw device is split into 4MB sized objects by default. + * The sequencenumber is encoded in a 12 byte long hex-string, + * and is attached to the devicename, separated by a dot. + * e.g. "devicename.1234567890ab" + * + */ + +#define OBJ_MAX_SIZE (1UL << OBJ_DEFAULT_OBJ_ORDER) + +typedef struct RBDAIOCB { + BlockDriverAIOCB common; + QEMUBH *bh; + int ret; + QEMUIOVector *qiov; + char *bounce; + int write; + int64_t sector_num; + int aiocnt; + int error; +} RBDAIOCB; + +typedef struct RADOSCB { + int rcbid; + RBDAIOCB *acb; + int done; + int64_t segsize; + char *buf; +} RADOSCB; + +typedef struct RBDRVRBDState { + rados_pool_t pool; + char name[RBD_MAX_OBJ_NAME_SIZE]; + int name_len; + uint64_t size; + uint64_t objsize; +} RBDRVRBDState; + +typedef struct rbd_obj_header_ondisk RbdHeader1; + +static int rbd_parsename(const char *filename, char *pool, char *name) +{ + const char *rbdname; + char *p, *n; + int l; + + if (!strstart(filename, "rbd:", &rbdname)) { + return -EINVAL; + } + + pstrcpy(pool, 2 * RBD_MAX_SEG_NAME_SIZE, rbdname); + p = strchr(pool, '/'); + if (p == NULL) { + return -EINVAL; + } + + *p = '\0'; + n = ++p; + + l = strlen(n); + + if (l > RBD_MAX_OBJ_NAME_SIZE) { + fprintf(stderr, "object name to long\n"); + return -EINVAL; + } else if (l <= 0) { + fprintf(stderr, "object name to short\n"); + return -EINVAL; + } + + strcpy(name, n); + + return l; +} + +static int create_tmap_op(uint8_t op, const char *name, char **tmap_desc) +{ + uint32_t len = strlen(name); + uint32_t total_len = 1 + (sizeof(uint32_t) + len) + sizeof(uint32_t); /* encoding op + name + empty buffer */ + char *desc; + + desc = qemu_malloc(total_len); + if (!desc) { + return -ENOMEM; + } + + *tmap_desc = desc; + + *desc = op; + desc++; + memcpy(desc, &len, sizeof(len)); + desc += sizeof(len); + memcpy(desc, name, len); + desc += len; + len = 0; + memcpy(desc, &len, sizeof(len)); + desc += sizeof(len); + + return desc - *tmap_desc; +} + +static void free_tmap_op(char *tmap_desc) +{ + qemu_free(tmap_desc); +} + +static int rbd_register_image(rados_pool_t pool, const char *name) +{ + char *tmap_desc; + const char *dir = RBD_DIRECTORY; + int ret; + + ret = create_tmap_op(CEPH_OSD_TMAP_SET, name, &tmap_desc); + if (ret < 0) { + return ret; + } + + ret = rados_tmap_update(pool, dir, tmap_desc, ret); + free_tmap_op(tmap_desc); + + return ret; +} + +static int rbd_create(const char *filename, QEMUOptionParameter *options) +{ + int64_t bytes = 0; + int64_t objsize; + uint64_t size; + time_t mtime; + uint8_t obj_order = RBD_DEFAULT_OBJ_ORDER; + char pool[RBD_MAX_SEG_NAME_SIZE]; + char n[RBD_MAX_SEG_NAME_SIZE]; + char name[RBD_MAX_SEG_NAME_SIZE]; + RbdHeader1 header; + rados_pool_t p; + int name_len; + int ret; + + if ((name_len = rbd_parsename(filename, pool, name)) < 0) { + return -EINVAL; + } + + snprintf(n, RBD_MAX_SEG_NAME_SIZE, "%s%s", name, RBD_SUFFIX); + + /* Read out options */ + while (options && options->name) { + if (!strcmp(options->name, BLOCK_OPT_SIZE)) { + bytes = options->value.n; + } else if (!strcmp(options->name, BLOCK_OPT_CLUSTER_SIZE)) { + if (options->value.n) { + objsize = options->value.n; + if (!objsize || ((objsize - 1) & objsize)) { /* not a power of 2? */ + fprintf(stderr, "obj size needs to be power of 2\n"); + return -EINVAL; + } + if (objsize < 4096) { + fprintf(stderr, "obj size too small\n"); + return -EINVAL; + } + + for (obj_order = 0; obj_order < 64; obj_order++) { + if (objsize == 1) + break; + objsize >>= 1; + } + } + } + options++; + } + + memset(&header, 0, sizeof(header)); + pstrcpy(header.text, sizeof(header.text), rbd_text); + pstrcpy(header.signature, sizeof(header.signature), rbd_signature); + pstrcpy(header.version, sizeof(header.version), rbd_version); + header.image_size = bytes; + cpu_to_le64s((uint64_t *) & header.image_size); + header.options.order = obj_order; + header.options.crypt_type = RBD_CRYPT_NONE; + header.options.comp_type = RBD_COMP_NONE; + header.snap_seq = 0; + header.snap_count = 0; + cpu_to_le32s(&header.snap_count); + + if (rados_initialize(0, NULL) < 0) { + fprintf(stderr, "error initializing\n"); + return -EIO; + } + + if (rados_open_pool(pool, &p)) { + fprintf(stderr, "error opening pool %s\n", pool); + return -EIO; + } + + /* check for existing rbd header file */ + ret = rados_stat(p, n, &size, &mtime); + if (ret == 0) { + ret=-EEXIST; + goto done; + } + + /* create header file */ + ret = rados_write(p, n, 0, (const char *)&header, sizeof(header)); + if (ret < 0) { + goto done; + } + + ret = rbd_register_image(p, name); +done: + rados_close_pool(p); + rados_deinitialize(); + + return ret; +} + +static int rbd_open(BlockDriverState *bs, const char *filename, int flags) +{ + RBDRVRBDState *s = bs->opaque; + char pool[RBD_MAX_SEG_NAME_SIZE]; + char n[RBD_MAX_SEG_NAME_SIZE]; + char hbuf[4096]; + + if ((s->name_len = rbd_parsename(filename, pool, s->name)) < 0) { + return -EINVAL; + } + snprintf(n, RBD_MAX_SEG_NAME_SIZE, "%s%s", s->name, RBD_SUFFIX); + + if (rados_initialize(0, NULL) < 0) { + fprintf(stderr, "error initializing\n"); + return -EIO; + } + + if (rados_open_pool(pool, &s->pool)) { + fprintf(stderr, "error opening pool %s\n", pool); + return -EIO; + } + + if (rados_read(s->pool, n, 0, hbuf, 4096) < 0) { + fprintf(stderr, "error reading header from %s\n", s->name); + return -EIO; + } + if (!strncmp(hbuf + 64, rbd_signature, 4)) { + if (!strncmp(hbuf + 68, rbd_version, 8)) { + RbdHeader1 *header; + + header = (RbdHeader1 *) hbuf; + le64_to_cpus((uint64_t *) & header->image_size); + s->size = header->image_size; + s->objsize = 1 << header->options.order; + } else { + fprintf(stderr, "Unknown image version %s\n", hbuf + 68); + return -EIO; + } + } else { + fprintf(stderr, "Invalid header signature %s\n", hbuf + 64); + return -EIO; + } + + return 0; +} + +static void rbd_close(BlockDriverState *bs) +{ + RBDRVRBDState *s = bs->opaque; + + rados_close_pool(s->pool); + rados_deinitialize(); +} + +static int rbd_rw(BlockDriverState *bs, int64_t sector_num, + uint8_t *buf, int nb_sectors, int write) +{ + RBDRVRBDState *s = bs->opaque; + char n[RBD_MAX_SEG_NAME_SIZE]; + + int64_t segnr, segoffs, segsize, r; + int64_t off, size; + + off = sector_num * 512; + size = nb_sectors * 512; + segnr = (int64_t) (off / s->objsize); + segoffs = (int64_t) (off % s->objsize); + segsize = (int64_t) (s->objsize - segoffs); + + while (size > 0) { + if (size < segsize) { + segsize = size; + } + + snprintf(n, RBD_MAX_SEG_NAME_SIZE, "%s.%012llx", s->name, + (long long unsigned int)segnr); + + if (write) { + if ((r = rados_write(s->pool, n, segoffs, (const char *)buf, + segsize)) < 0) { + return r; + } + } else { + r = rados_read(s->pool, n, segoffs, (char *)buf, segsize); + if (r == -ENOENT) { + memset(buf, 0, segsize); + } else if (r < 0) { + return(r); + } else if (r < segsize) { + memset(buf + r, 0, segsize - r); + } + r = segsize; + } + + buf += segsize; + size -= segsize; + segoffs = 0; + segsize = s->objsize; + segnr++; + } + + return (0); +} + +static int rbd_read(BlockDriverState *bs, int64_t sector_num, + uint8_t *buf, int nb_sectors) +{ + return rbd_rw(bs, sector_num, buf, nb_sectors, 0); +} + +static int rbd_write(BlockDriverState *bs, int64_t sector_num, + const uint8_t *buf, int nb_sectors) +{ + return rbd_rw(bs, sector_num, (uint8_t *) buf, nb_sectors, 1); +} + +static void rbd_aio_cancel(BlockDriverAIOCB *blockacb) +{ + RBDAIOCB *acb = (RBDAIOCB *) blockacb; + qemu_bh_delete(acb->bh); + acb->bh = NULL; + qemu_aio_release(acb); +} + +static AIOPool rbd_aio_pool = { + .aiocb_size = sizeof(RBDAIOCB), + .cancel = rbd_aio_cancel, +}; + +/* This is the callback function for rados_aio_read and _write */ +static void rbd_finish_aiocb(rados_completion_t c, RADOSCB *rcb) +{ + RBDAIOCB *acb = rcb->acb; + int64_t r; + int i; + + acb->aiocnt--; + r = rados_aio_get_return_value(c); + rados_aio_release(c); + if (acb->write) { + if (r < 0) { + acb->ret = r; + acb->error = 1; + } else if (!acb->error) { + acb->ret += rcb->segsize; + } + } else { + if (r == -ENOENT) { + memset(rcb->buf, 0, rcb->segsize); + if (!acb->error) { + acb->ret += rcb->segsize; + } + } else if (r < 0) { + acb->ret = r; + acb->error = 1; + } else if (r < rcb->segsize) { + memset(rcb->buf + r, 0, rcb->segsize - r); + if (!acb->error) { + acb->ret += rcb->segsize; + } + } else if (!acb->error) { + acb->ret += r; + } + } + qemu_free(rcb); + i = 0; + if (!acb->aiocnt && acb->bh) { + qemu_bh_schedule(acb->bh); + } +} + +/* Callback when all queued rados_aio requests are complete */ +static void rbd_aio_bh_cb(void *opaque) +{ + RBDAIOCB *acb = opaque; + + if (!acb->write) { + qemu_iovec_from_buffer(acb->qiov, acb->bounce, acb->qiov->size); + } + qemu_vfree(acb->bounce); + acb->common.cb(acb->common.opaque, (acb->ret > 0 ? 0 : acb->ret)); + qemu_bh_delete(acb->bh); + acb->bh = NULL; + qemu_aio_release(acb); +} + +static BlockDriverAIOCB *rbd_aio_rw_vector(BlockDriverState *bs, + int64_t sector_num, + QEMUIOVector *qiov, + int nb_sectors, + BlockDriverCompletionFunc *cb, + void *opaque, int write) +{ + RBDAIOCB *acb; + RADOSCB *rcb; + rados_completion_t c; + char n[RBD_MAX_SEG_NAME_SIZE]; + int64_t segnr, segoffs, segsize, last_segnr; + int64_t off, size; + char *buf; + + RBDRVRBDState *s = bs->opaque; + + acb = qemu_aio_get(&rbd_aio_pool, bs, cb, opaque); + acb->write = write; + acb->qiov = qiov; + acb->bounce = qemu_blockalign(bs, qiov->size); + acb->aiocnt = 0; + acb->ret = 0; + acb->error = 0; + + if (!acb->bh) { + acb->bh = qemu_bh_new(rbd_aio_bh_cb, acb); + } + + if (write) { + qemu_iovec_to_buffer(acb->qiov, acb->bounce); + } + + buf = acb->bounce; + + off = sector_num * 512; + size = nb_sectors * 512; + segnr = (int64_t) (off / s->objsize); + segoffs = (int64_t) (off % s->objsize); + segsize = (int64_t) (s->objsize - segoffs); + + last_segnr = ((off + size - 1) / s->objsize); + acb->aiocnt = (last_segnr - segnr) + 1; + + while (size > 0) { + if (size < segsize) { + segsize = size; + } + + snprintf(n, RBD_MAX_SEG_NAME_SIZE, "%s.%012llx", s->name, + (long long unsigned int)segnr); + + rcb = qemu_malloc(sizeof(RADOSCB)); + rcb->done = 0; + rcb->acb = acb; + rcb->segsize = segsize; + rcb->buf = buf; + + if (write) { + rados_aio_create_completion(rcb, NULL, + (rados_callback_t) rbd_finish_aiocb, &c); + rados_aio_write(s->pool, n, segoffs, buf, segsize, c); + } else { + rados_aio_create_completion(rcb, (rados_callback_t) rbd_finish_aiocb, + NULL, &c); + rados_aio_read(s->pool, n, segoffs, buf, segsize, c); + } + + buf += segsize; + size -= segsize; + segoffs = 0; + segsize = s->objsize; + segnr++; + } + + return &acb->common; +} + +static BlockDriverAIOCB *rbd_aio_readv(BlockDriverState * bs, + int64_t sector_num, QEMUIOVector * qiov, + int nb_sectors, + BlockDriverCompletionFunc * cb, + void *opaque) +{ + return rbd_aio_rw_vector(bs, sector_num, qiov, nb_sectors, cb, opaque, 0); +} + +static BlockDriverAIOCB *rbd_aio_writev(BlockDriverState * bs, + int64_t sector_num, QEMUIOVector * qiov, + int nb_sectors, + BlockDriverCompletionFunc * cb, + void *opaque) +{ + return rbd_aio_rw_vector(bs, sector_num, qiov, nb_sectors, cb, opaque, 1); +} + +static int rbd_getinfo(BlockDriverState * bs, BlockDriverInfo * bdi) +{ + RBDRVRBDState *s = bs->opaque; + bdi->cluster_size = s->objsize; + return 0; +} + +static int64_t rbd_getlength(BlockDriverState * bs) +{ + RBDRVRBDState *s = bs->opaque; + + return s->size; +} + +static QEMUOptionParameter rbd_create_options[] = { + { + .name = BLOCK_OPT_SIZE, + .type = OPT_SIZE, + .help = "Virtual disk size" + }, + { + .name = BLOCK_OPT_CLUSTER_SIZE, + .type = OPT_SIZE, + .help = "RBD object size" + }, + {NULL} +}; + +static BlockDriver bdrv_rbd = { + .format_name = "rbd", + .instance_size = sizeof(RBDRVRBDState), + .bdrv_file_open = rbd_open, + .bdrv_read = rbd_read, + .bdrv_write = rbd_write, + .bdrv_close = rbd_close, + .bdrv_create = rbd_create, + .bdrv_get_info = rbd_getinfo, + .create_options = rbd_create_options, + .bdrv_getlength = rbd_getlength, + .protocol_name = "rbd", + + .bdrv_aio_readv = rbd_aio_readv, + .bdrv_aio_writev = rbd_aio_writev, +}; + +static void bdrv_rbd_init(void) +{ + bdrv_register(&bdrv_rbd); +} + +block_init(bdrv_rbd_init); diff --git a/block/rbd_types.h b/block/rbd_types.h new file mode 100644 index 0000000..3a16a26 --- /dev/null +++ b/block/rbd_types.h @@ -0,0 +1,52 @@ +#ifndef QEMU_BLOCK_RBD_TYPES_H +#define QEMU_BLOCK_RBD_TYPES_H + + +/* + * rbd image 'foo' consists of objects + * foo.rbd - image metadata + * foo.00000000 + * foo.00000001 + * ... - data + */ + +#define RBD_SUFFIX ".rbd" +#define RBD_DIRECTORY "rbd_directory" + +#define RBD_DEFAULT_OBJ_ORDER 22 /* 4MB */ + +#define RBD_MAX_OBJ_NAME_SIZE 96 +#define RBD_MAX_SEG_NAME_SIZE 128 + +#define RBD_COMP_NONE 0 +#define RBD_CRYPT_NONE 0 + +static const char rbd_text[] = "<<< Rados Block Device Image >>>\n"; +static const char rbd_signature[] = "RBD"; +static const char rbd_version[] = "001.004"; + +struct rbd_obj_snap_ondisk { + uint64_t id; + uint64_t image_size; +} __attribute__((packed)); + +struct rbd_obj_header_ondisk { + char text[64]; + char signature[4]; + char version[8]; + struct { + uint8_t order; + uint8_t crypt_type; + uint8_t comp_type; + uint8_t unused; + } __attribute__((packed)) options; + uint64_t image_size; + uint64_t snap_seq; + uint32_t snap_count; + uint32_t reserved; + uint64_t snap_names_len; + struct rbd_obj_snap_ondisk snaps[0]; +} __attribute__((packed)); + + +#endif diff --git a/configure b/configure index 3cd2c5f..7d70bf8 100755 --- a/configure +++ b/configure @@ -299,6 +299,7 @@ pkgversion="" check_utests="no" user_pie="no" zero_malloc="" +rbd="no" # OS specific if check_define __linux__ ; then @@ -660,6 +661,8 @@ for opt do ;; --enable-vhost-net) vhost_net="yes" ;; + --enable-rbd) rbd="yes" + ;; *) echo "ERROR: unknown option $opt"; show_help="yes" ;; esac @@ -826,6 +829,7 @@ echo " --enable-docs enable documentation build" echo " --disable-docs disable documentation build" echo " --disable-vhost-net disable vhost-net acceleration support" echo " --enable-vhost-net enable vhost-net acceleration support" +echo " --enable-rbd enable building the rados block device (rbd)" echo "" echo "NOTE: The object files are built at the place where configure is launched" exit 1 @@ -1579,6 +1583,25 @@ if test "$mingw32" != yes -a "$pthread" = no; then fi ########################################## +# rbd probe +if test "$rbd" != "no" ; then + cat > $TMPC <<EOF +#include <stdio.h> +#include <rados/librados.h> +int main(void) { rados_initialize(0, NULL); return 0; } +EOF + if compile_prog "" "-lrados -lcrypto" ; then + rbd=yes + LIBS="$LIBS -lrados -lcrypto" + else + if test "$rbd" = "yes" ; then + feature_not_found "rados block device" + fi + rbd=no + fi +fi + +########################################## # linux-aio probe if test "$linux_aio" != "no" ; then @@ -2041,6 +2064,7 @@ echo "preadv support $preadv" echo "fdatasync $fdatasync" echo "uuid support $uuid" echo "vhost-net support $vhost_net" +echo "rbd support $rbd" if test $sdl_too_old = "yes"; then echo "-> Your SDL version is too old - please upgrade to have SDL support" @@ -2270,6 +2294,9 @@ echo "CONFIG_UNAME_RELEASE=\"$uname_release\"" >> $config_host_mak if test "$zero_malloc" = "yes" ; then echo "CONFIG_ZERO_MALLOC=y" >> $config_host_mak fi +if test "$rbd" = "yes" ; then + echo "CONFIG_RBD=y" >> $config_host_mak +fi # USB host support case "$usb" in