Apart from two cosmetic issues (see below), I think this patch is ready to replace the old rbd driver. You can add:
Reviewed-by: Christian Brunner <c...@muc.de> Regards Christian 2011/5/24 Josh Durgin <josh.dur...@dreamhost.com>: > librbd stacks on top of librados to provide access > to rbd images. > > Using librbd simplifies the qemu code, and allows > qemu to use new versions of the rbd format > with few (if any) changes. > > Signed-off-by: Josh Durgin <josh.dur...@dreamhost.com> > Signed-off-by: Yehuda Sadeh <yeh...@hq.newdream.net> > --- > block/rbd.c | 790 > +++++++++++++++-------------------------------------- > block/rbd_types.h | 71 ----- > configure | 33 +-- > 3 files changed, 224 insertions(+), 670 deletions(-) > delete mode 100644 block/rbd_types.h > > diff --git a/block/rbd.c b/block/rbd.c > index 249a590..1c8e7c7 100644 > --- a/block/rbd.c > +++ b/block/rbd.c > @@ -1,20 +1,22 @@ > /* > * QEMU Block driver for RADOS (Ceph) > * > - * Copyright (C) 2010 Christian Brunner <c...@muc.de> > + * Copyright (C) 2010-2011 Christian Brunner <c...@muc.de>, > + * Josh Durgin <josh.dur...@dreamhost.com> > * > * This work is licensed under the terms of the GNU GPL, version 2. See > * the COPYING file in the top-level directory. > * > */ > > +#include <inttypes.h> > + > #include "qemu-common.h" > #include "qemu-error.h" > > -#include "rbd_types.h" > #include "block_int.h" > > -#include <rados/librados.h> > +#include <rbd/librbd.h> > > > > @@ -40,6 +42,12 @@ > > #define OBJ_MAX_SIZE (1UL << OBJ_DEFAULT_OBJ_ORDER) > > +#define RBD_MAX_CONF_NAME_SIZE 128 > +#define RBD_MAX_CONF_VAL_SIZE 512 > +#define RBD_MAX_CONF_SIZE 1024 > +#define RBD_MAX_POOL_NAME_SIZE 128 > +#define RBD_MAX_SNAP_NAME_SIZE 128 > + > typedef struct RBDAIOCB { > BlockDriverAIOCB common; > QEMUBH *bh; > @@ -48,7 +56,6 @@ typedef struct RBDAIOCB { > char *bounce; > int write; > int64_t sector_num; > - int aiocnt; > int error; > struct BDRVRBDState *s; > int cancelled; > @@ -59,7 +66,7 @@ typedef struct RADOSCB { > RBDAIOCB *acb; > struct BDRVRBDState *s; > int done; > - int64_t segsize; > + int64_t size; > char *buf; > int ret; > } RADOSCB; > @@ -69,25 +76,22 @@ typedef struct RADOSCB { > > typedef struct BDRVRBDState { > int fds[2]; > - rados_pool_t pool; > - rados_pool_t header_pool; > - char name[RBD_MAX_OBJ_NAME_SIZE]; > - char block_name[RBD_MAX_BLOCK_NAME_SIZE]; > - uint64_t size; > - uint64_t objsize; > + rados_t cluster; > + rados_ioctx_t io_ctx; > + rbd_image_t image; > + char name[RBD_MAX_IMAGE_NAME_SIZE]; > int qemu_aio_count; > + char *snap; > int event_reader_pos; > RADOSCB *event_rcb; > } BDRVRBDState; > > -typedef struct rbd_obj_header_ondisk RbdHeader1; > - > static void rbd_aio_bh_cb(void *opaque); > > -static int rbd_next_tok(char *dst, int dst_len, > - char *src, char delim, > - const char *name, > - char **p) > +static int qemu_rbd_next_tok(char *dst, int dst_len, > + char *src, char delim, > + const char *name, > + char **p) > { > int l; > char *end; > @@ -115,10 +119,10 @@ static int rbd_next_tok(char *dst, int dst_len, > return 0; > } > > -static int rbd_parsename(const char *filename, > - char *pool, int pool_len, > - char *snap, int snap_len, > - char *name, int name_len) > +static int qemu_rbd_parsename(const char *filename, > + char *pool, int pool_len, > + char *snap, int snap_len, > + char *name, int name_len) > { > const char *start; > char *p, *buf; > @@ -131,12 +135,12 @@ static int rbd_parsename(const char *filename, > buf = qemu_strdup(start); > p = buf; > > - ret = rbd_next_tok(pool, pool_len, p, '/', "pool name", &p); > + ret = qemu_rbd_next_tok(pool, pool_len, p, '/', "pool name", &p); > if (ret < 0 || !p) { > ret = -EINVAL; > goto done; > } > - ret = rbd_next_tok(name, name_len, p, '@', "object name", &p); > + ret = qemu_rbd_next_tok(name, name_len, p, '@', "object name", &p); > if (ret < 0) { > goto done; > } > @@ -145,123 +149,35 @@ static int rbd_parsename(const char *filename, > goto done; > } > > - ret = rbd_next_tok(snap, snap_len, p, '\0', "snap name", &p); > + ret = qemu_rbd_next_tok(snap, snap_len, p, '\0', "snap name", &p); > > done: > qemu_free(buf); > return ret; > } > > -static int create_tmap_op(uint8_t op, const char *name, char **tmap_desc) > -{ > - uint32_t len = strlen(name); > - uint32_t len_le = cpu_to_le32(len); > - /* total_len = encoding op + name + empty buffer */ > - uint32_t total_len = 1 + (sizeof(uint32_t) + len) + sizeof(uint32_t); > - uint8_t *desc = NULL; > - > - desc = qemu_malloc(total_len); > - > - *tmap_desc = (char *)desc; > - > - *desc = op; > - desc++; > - memcpy(desc, &len_le, sizeof(len_le)); > - desc += sizeof(len_le); > - memcpy(desc, name, len); > - desc += len; > - len = 0; /* no need for endian conversion for 0 */ > - memcpy(desc, &len, sizeof(len)); > - desc += sizeof(len); > - > - return (char *)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 touch_rbd_info(rados_pool_t pool, const char *info_oid) > -{ > - int r = rados_write(pool, info_oid, 0, NULL, 0); > - if (r < 0) { > - return r; > - } > - return 0; > -} > - > -static int rbd_assign_bid(rados_pool_t pool, uint64_t *id) > -{ > - uint64_t out[1]; > - const char *info_oid = RBD_INFO; > - > - *id = 0; > - > - int r = touch_rbd_info(pool, info_oid); > - if (r < 0) { > - return r; > - } > - > - r = rados_exec(pool, info_oid, "rbd", "assign_bid", NULL, > - 0, (char *)out, sizeof(out)); > - if (r < 0) { > - return r; > - } > - > - le64_to_cpus(out); > - *id = out[0]; > - > - return 0; > -} > - > -static int rbd_create(const char *filename, QEMUOptionParameter *options) > +static int qemu_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_OBJ_NAME_SIZE]; > - char snap_buf[RBD_MAX_SEG_NAME_SIZE]; > + int obj_order = 0; > + char pool[RBD_MAX_POOL_NAME_SIZE]; > + char name[RBD_MAX_IMAGE_NAME_SIZE]; > + char snap_buf[RBD_MAX_SNAP_NAME_SIZE]; > char *snap = NULL; > - RbdHeader1 header; > - rados_pool_t p; > - uint64_t bid; > - uint32_t hi, lo; > + rados_t cluster; > + rados_ioctx_t io_ctx; > int ret; > > - if (rbd_parsename(filename, > - pool, sizeof(pool), > - snap_buf, sizeof(snap_buf), > - name, sizeof(name)) < 0) { > + if (qemu_rbd_parsename(filename, pool, sizeof(pool), > + snap_buf, sizeof(snap_buf), > + name, sizeof(name)) < 0) { > return -EINVAL; > } > if (snap_buf[0] != '\0') { > snap = snap_buf; > } > > - snprintf(n, sizeof(n), "%s%s", name, RBD_SUFFIX); > - > /* Read out options */ > while (options && options->name) { > if (!strcmp(options->name, BLOCK_OPT_SIZE)) { > @@ -277,82 +193,55 @@ static int rbd_create(const char *filename, > QEMUOptionParameter *options) > error_report("obj size too small"); > return -EINVAL; > } > - obj_order = ffs(objsize) - 1; > + obj_order = ffs(objsize) - 1; > } > } > options++; > } > > - memset(&header, 0, sizeof(header)); > - pstrcpy(header.text, sizeof(header.text), RBD_HEADER_TEXT); > - pstrcpy(header.signature, sizeof(header.signature), > RBD_HEADER_SIGNATURE); > - pstrcpy(header.version, sizeof(header.version), RBD_HEADER_VERSION); > - header.image_size = cpu_to_le64(bytes); > - 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; > - > - if (rados_initialize(0, NULL) < 0) { > + if (rados_create(&cluster, NULL) < 0) { > error_report("error initializing"); > return -EIO; > } > > - if (rados_open_pool(pool, &p)) { > - error_report("error opening pool %s", pool); > - rados_deinitialize(); > + if (rados_conf_read_file(cluster, NULL) < 0) { > + error_report("error reading config file"); > + rados_shutdown(cluster); > return -EIO; > } > > - /* check for existing rbd header file */ > - ret = rados_stat(p, n, &size, &mtime); > - if (ret == 0) { > - ret=-EEXIST; > - goto done; > - } > - > - ret = rbd_assign_bid(p, &bid); > - if (ret < 0) { > - error_report("failed assigning block id"); > - rados_deinitialize(); > + if (rados_connect(cluster) < 0) { > + error_report("error connecting"); > + rados_shutdown(cluster); > return -EIO; > } > - hi = bid >> 32; > - lo = bid & 0xFFFFFFFF; > - snprintf(header.block_name, sizeof(header.block_name), "rb.%x.%x", hi, > lo); > > - /* create header file */ > - ret = rados_write(p, n, 0, (const char *)&header, sizeof(header)); > - if (ret < 0) { > - goto done; > + if (rados_ioctx_create(cluster, pool, &io_ctx) < 0) { > + error_report("error opening pool %s", pool); > + rados_shutdown(cluster); > + return -EIO; > } > > - ret = rbd_register_image(p, name); > -done: > - rados_close_pool(p); > - rados_deinitialize(); > + ret = rbd_create(io_ctx, name, bytes, &obj_order); > + rados_ioctx_destroy(io_ctx); > + rados_shutdown(cluster); > > return ret; > } > > /* > - * This aio completion is being called from rbd_aio_event_reader() and > - * runs in qemu context. It schedules a bh, but just in case the aio > + * This aio completion is being called from qemu_rbd_aio_event_reader() > + * and runs in qemu context. It schedules a bh, but just in case the aio > * was not cancelled before. > */ > -static void rbd_complete_aio(RADOSCB *rcb) > +static void qemu_rbd_complete_aio(RADOSCB *rcb) > { > RBDAIOCB *acb = rcb->acb; > int64_t r; > > - acb->aiocnt--; > - > if (acb->cancelled) { > - if (!acb->aiocnt) { > - qemu_vfree(acb->bounce); > - qemu_aio_release(acb); > - } > + qemu_vfree(acb->bounce); > + qemu_aio_release(acb); > goto done; > } > > @@ -363,32 +252,25 @@ static void rbd_complete_aio(RADOSCB *rcb) > acb->ret = r; > acb->error = 1; > } else if (!acb->error) { > - acb->ret += rcb->segsize; > + acb->ret = rcb->size; > } > } else { > - if (r == -ENOENT) { > - memset(rcb->buf, 0, rcb->segsize); > - if (!acb->error) { > - acb->ret += rcb->segsize; > - } > - } else if (r < 0) { > - memset(rcb->buf, 0, rcb->segsize); > + if (r < 0) { > + memset(rcb->buf, 0, rcb->size); > acb->ret = r; > acb->error = 1; > - } else if (r < rcb->segsize) { > - memset(rcb->buf + r, 0, rcb->segsize - r); > + } else if (r < rcb->size) { > + memset(rcb->buf + r, 0, rcb->size - r); > if (!acb->error) { > - acb->ret += rcb->segsize; > + acb->ret = rcb->size; > } > } else if (!acb->error) { > - acb->ret += r; > + acb->ret = r; > } > } > /* Note that acb->bh can be NULL in case where the aio was cancelled */ > - if (!acb->aiocnt) { > - acb->bh = qemu_bh_new(rbd_aio_bh_cb, acb); > - qemu_bh_schedule(acb->bh); > - } > + acb->bh = qemu_bh_new(rbd_aio_bh_cb, acb); > + qemu_bh_schedule(acb->bh); > done: > qemu_free(rcb); > } > @@ -397,7 +279,7 @@ done: > * aio fd read handler. It runs in the qemu context and calls the > * completion handling of completed rados aio operations. > */ > -static void rbd_aio_event_reader(void *opaque) > +static void qemu_rbd_aio_event_reader(void *opaque) > { > BDRVRBDState *s = opaque; > > @@ -413,176 +295,74 @@ static void rbd_aio_event_reader(void *opaque) > s->event_reader_pos += ret; > if (s->event_reader_pos == sizeof(s->event_rcb)) { > s->event_reader_pos = 0; > - rbd_complete_aio(s->event_rcb); > - s->qemu_aio_count --; > + qemu_rbd_complete_aio(s->event_rcb); > + s->qemu_aio_count--; > } > } > } > } while (ret < 0 && errno == EINTR); > } > > -static int rbd_aio_flush_cb(void *opaque) > +static int qemu_rbd_aio_flush_cb(void *opaque) > { > BDRVRBDState *s = opaque; > > return (s->qemu_aio_count > 0); > } > > - > -static int rbd_set_snapc(rados_pool_t pool, const char *snap, RbdHeader1 > *header) > -{ > - uint32_t snap_count = le32_to_cpu(header->snap_count); > - rados_snap_t *snaps = NULL; > - rados_snap_t seq; > - uint32_t i; > - uint64_t snap_names_len = le64_to_cpu(header->snap_names_len); > - int r; > - rados_snap_t snapid = 0; > - > - if (snap_count) { > - const char *header_snap = (const char *)&header->snaps[snap_count]; > - const char *end = header_snap + snap_names_len; > - snaps = qemu_malloc(sizeof(rados_snap_t) * header->snap_count); > - > - for (i=0; i < snap_count; i++) { > - snaps[i] = le64_to_cpu(header->snaps[i].id); > - > - if (snap && strcmp(snap, header_snap) == 0) { > - snapid = snaps[i]; > - } > - > - header_snap += strlen(header_snap) + 1; > - if (header_snap > end) { > - error_report("bad header, snapshot list broken"); > - } > - } > - } > - > - if (snap && !snapid) { > - error_report("snapshot not found"); > - qemu_free(snaps); > - return -ENOENT; > - } > - seq = le32_to_cpu(header->snap_seq); > - > - r = rados_set_snap_context(pool, seq, snaps, snap_count); > - > - rados_set_snap(pool, snapid); > - > - qemu_free(snaps); > - > - return r; > -} > - > -#define BUF_READ_START_LEN 4096 > - > -static int rbd_read_header(BDRVRBDState *s, char **hbuf) > -{ > - char *buf = NULL; > - char n[RBD_MAX_SEG_NAME_SIZE]; > - uint64_t len = BUF_READ_START_LEN; > - int r; > - > - snprintf(n, sizeof(n), "%s%s", s->name, RBD_SUFFIX); > - > - buf = qemu_malloc(len); > - > - r = rados_read(s->header_pool, n, 0, buf, len); > - if (r < 0) { > - goto failed; > - } > - > - if (r < len) { > - goto done; > - } > - > - qemu_free(buf); > - buf = qemu_malloc(len); > - > - r = rados_stat(s->header_pool, n, &len, NULL); > - if (r < 0) { > - goto failed; > - } > - > - r = rados_read(s->header_pool, n, 0, buf, len); > - if (r < 0) { > - goto failed; > - } > - > -done: > - *hbuf = buf; > - return 0; > - > -failed: > - qemu_free(buf); > - return r; > -} > - > -static int rbd_open(BlockDriverState *bs, const char *filename, int flags) > +static int qemu_rbd_open(BlockDriverState *bs, const char *filename, int > flags) > { > BDRVRBDState *s = bs->opaque; > - RbdHeader1 *header; > - char pool[RBD_MAX_SEG_NAME_SIZE]; > - char snap_buf[RBD_MAX_SEG_NAME_SIZE]; > - char *snap = NULL; > - char *hbuf = NULL; > + char pool[RBD_MAX_POOL_NAME_SIZE]; > + char snap_buf[RBD_MAX_SNAP_NAME_SIZE]; > int r; > > - if (rbd_parsename(filename, pool, sizeof(pool), > - snap_buf, sizeof(snap_buf), > - s->name, sizeof(s->name)) < 0) { > + if (qemu_rbd_parsename(filename, pool, sizeof(pool), > + snap_buf, sizeof(snap_buf), > + s->name, sizeof(s->name)) < 0) { > return -EINVAL; > } > + s->snap = NULL; > if (snap_buf[0] != '\0') { > - snap = snap_buf; > + s->snap = qemu_strdup(snap_buf); > } > > - if ((r = rados_initialize(0, NULL)) < 0) { > + r = rados_create(&s->cluster, NULL); > + if (r < 0) { > error_report("error initializing"); > return r; > } > > - if ((r = rados_open_pool(pool, &s->pool))) { > - error_report("error opening pool %s", pool); > - rados_deinitialize(); > + r = rados_conf_read_file(s->cluster, NULL); > + if (r < 0) { > + error_report("error reading config file"); > + rados_shutdown(s->cluster); > return r; > } > > - if ((r = rados_open_pool(pool, &s->header_pool))) { > - error_report("error opening pool %s", pool); > - rados_deinitialize(); > + r = rados_connect(s->cluster); > + if (r < 0) { > + error_report("error connecting"); > + rados_shutdown(s->cluster); > return r; > } > > - if ((r = rbd_read_header(s, &hbuf)) < 0) { > - error_report("error reading header from %s", s->name); > - goto failed; > - } > - > - if (memcmp(hbuf + 64, RBD_HEADER_SIGNATURE, 4)) { > - error_report("Invalid header signature"); > - r = -EMEDIUMTYPE; > - goto failed; > - } > - > - if (memcmp(hbuf + 68, RBD_HEADER_VERSION, 8)) { > - error_report("Unknown image version"); > - r = -EMEDIUMTYPE; > - goto failed; > + r = rados_ioctx_create(s->cluster, pool, &s->io_ctx); > + if (r < 0) { > + error_report("error opening pool %s", pool); > + rados_shutdown(s->cluster); > + return r; > } > > - header = (RbdHeader1 *) hbuf; > - s->size = le64_to_cpu(header->image_size); > - s->objsize = 1ULL << header->options.order; > - memcpy(s->block_name, header->block_name, sizeof(header->block_name)); > - > - r = rbd_set_snapc(s->pool, snap, header); > + r = rbd_open(s->io_ctx, s->name, &s->image, s->snap); > if (r < 0) { > - error_report("failed setting snap context: %s", strerror(-r)); > - goto failed; > + error_report("error reading header from %s", s->name); > + rados_ioctx_destroy(s->io_ctx); > + rados_shutdown(s->cluster); > + return r; > } > > - bs->read_only = (snap != NULL); > + bs->read_only = (s->snap != NULL); > > s->event_reader_pos = 0; > r = qemu_pipe(s->fds); > @@ -592,23 +372,20 @@ static int rbd_open(BlockDriverState *bs, const char > *filename, int flags) > } > fcntl(s->fds[0], F_SETFL, O_NONBLOCK); > fcntl(s->fds[1], F_SETFL, O_NONBLOCK); > - qemu_aio_set_fd_handler(s->fds[RBD_FD_READ], rbd_aio_event_reader, NULL, > - rbd_aio_flush_cb, NULL, s); > + qemu_aio_set_fd_handler(s->fds[RBD_FD_READ], qemu_rbd_aio_event_reader, > + NULL, qemu_rbd_aio_flush_cb, NULL, s); > > - qemu_free(hbuf); > > return 0; > > failed: > - qemu_free(hbuf); > - > - rados_close_pool(s->header_pool); > - rados_close_pool(s->pool); > - rados_deinitialize(); > + rbd_close(s->image); > + rados_ioctx_destroy(s->io_ctx); > + rados_shutdown(s->cluster); > return r; > } > > -static void rbd_close(BlockDriverState *bs) > +static void qemu_rbd_close(BlockDriverState *bs) > { > BDRVRBDState *s = bs->opaque; > > @@ -617,16 +394,17 @@ static void rbd_close(BlockDriverState *bs) > qemu_aio_set_fd_handler(s->fds[RBD_FD_READ], NULL , NULL, NULL, NULL, > NULL); > > - rados_close_pool(s->header_pool); > - rados_close_pool(s->pool); > - rados_deinitialize(); > + rbd_close(s->image); > + rados_ioctx_destroy(s->io_ctx); > + qemu_free(s->snap); > + rados_shutdown(s->cluster); > } > > /* > * Cancel aio. Since we don't reference acb in a non qemu threads, > * it is safe to access it here. > */ > -static void rbd_aio_cancel(BlockDriverAIOCB *blockacb) > +static void qemu_rbd_aio_cancel(BlockDriverAIOCB *blockacb) > { > RBDAIOCB *acb = (RBDAIOCB *) blockacb; > acb->cancelled = 1; > @@ -634,39 +412,28 @@ static void rbd_aio_cancel(BlockDriverAIOCB *blockacb) > > static AIOPool rbd_aio_pool = { > .aiocb_size = sizeof(RBDAIOCB), > - .cancel = rbd_aio_cancel, > + .cancel = qemu_rbd_aio_cancel, > }; > > -/* > - * This is the callback function for rados_aio_read and _write > - * > - * Note: this function is being called from a non qemu thread so > - * we need to be careful about what we do here. Generally we only > - * write to the block notification pipe, and do the rest of the > - * io completion handling from rbd_aio_event_reader() which > - * runs in a qemu context. > - */ > -static void rbd_finish_aiocb(rados_completion_t c, RADOSCB *rcb) > +static int qemu_rbd_send_pipe(BDRVRBDState *s, RADOSCB *rcb) > { > - int ret; > - rcb->ret = rados_aio_get_return_value(c); > - rados_aio_release(c); > + int ret = 0; > while (1) { > fd_set wfd; > - int fd = rcb->s->fds[RBD_FD_WRITE]; > + int fd = s->fds[RBD_FD_WRITE]; > > - /* send the rcb pointer to the qemu thread that is responsible > - for the aio completion. Must do it in a qemu thread context */ > + /* send the op pointer to the qemu thread that is responsible > + for the aio/op completion. Must do it in a qemu thread context */ > ret = write(fd, (void *)&rcb, sizeof(rcb)); > if (ret >= 0) { > break; > } > if (errno == EINTR) { > continue; > - } > + } > if (errno != EAGAIN) { > break; > - } > + } > > FD_ZERO(&wfd); > FD_SET(fd, &wfd); > @@ -675,13 +442,31 @@ static void rbd_finish_aiocb(rados_completion_t c, > RADOSCB *rcb) > } while (ret < 0 && errno == EINTR); > } > > + return ret; > +} > + > +/* > + * This is the callback function for rbd_aio_read and _write > + * > + * Note: this function is being called from a non qemu thread so > + * we need to be careful about what we do here. Generally we only > + * write to the block notification pipe, and do the rest of the > + * io completion handling from qemu_rbd_aio_event_reader() which > + * runs in a qemu context. > + */ > +static void rbd_finish_aiocb(rbd_completion_t c, RADOSCB *rcb) > +{ > + int ret; > + rcb->ret = rbd_aio_get_return_value(c); > + rbd_aio_release(c); > + ret = qemu_rbd_send_pipe(rcb->s, rcb); > if (ret < 0) { > - error_report("failed writing to acb->s->fds\n"); > + error_report("failed writing to acb->s->fds"); > qemu_free(rcb); > } > } > > -/* Callback when all queued rados_aio requests are complete */ > +/* Callback when all queued rbd_aio requests are complete */ > > static void rbd_aio_bh_cb(void *opaque) > { > @@ -707,9 +492,7 @@ static BlockDriverAIOCB > *rbd_aio_rw_vector(BlockDriverState *bs, > { > RBDAIOCB *acb; > RADOSCB *rcb; > - rados_completion_t c; > - char n[RBD_MAX_SEG_NAME_SIZE]; > - int64_t segnr, segoffs, segsize, last_segnr; > + rbd_completion_t c; > int64_t off, size; > char *buf; > > @@ -719,7 +502,6 @@ static BlockDriverAIOCB > *rbd_aio_rw_vector(BlockDriverState *bs, > acb->write = write; > acb->qiov = qiov; > acb->bounce = qemu_blockalign(bs, qiov->size); > - acb->aiocnt = 0; > acb->ret = 0; > acb->error = 0; > acb->s = s; > @@ -734,95 +516,81 @@ static BlockDriverAIOCB > *rbd_aio_rw_vector(BlockDriverState *bs, > > off = sector_num * BDRV_SECTOR_SIZE; > size = nb_sectors * BDRV_SECTOR_SIZE; > - segnr = off / s->objsize; > - segoffs = off % s->objsize; > - segsize = s->objsize - segoffs; > - > - last_segnr = ((off + size - 1) / s->objsize); > - acb->aiocnt = (last_segnr - segnr) + 1; > > - s->qemu_aio_count += acb->aiocnt; /* All the RADOSCB */ > + s->qemu_aio_count++; /* All the RADOSCB */ > > - while (size > 0) { > - if (size < segsize) { > - segsize = size; > - } > - > - snprintf(n, sizeof(n), "%s.%012" PRIx64, s->block_name, > - segnr); > - > - rcb = qemu_malloc(sizeof(RADOSCB)); > - rcb->done = 0; > - rcb->acb = acb; > - rcb->segsize = segsize; > - rcb->buf = buf; > - rcb->s = acb->s; > - > - 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); > - } > + rcb = qemu_malloc(sizeof(RADOSCB)); > + rcb->done = 0; > + rcb->acb = acb; > + rcb->buf = buf; > + rcb->s = acb->s; > + rcb->size = size; > > - buf += segsize; > - size -= segsize; > - segoffs = 0; > - segsize = s->objsize; > - segnr++; > + if (write) { > + rbd_aio_create_completion(rcb, (rbd_callback_t) rbd_finish_aiocb, > &c); > + rbd_aio_write(s->image, off, size, buf, c); > + } else { > + rbd_aio_create_completion(rcb, (rbd_callback_t) rbd_finish_aiocb, > &c); > + rbd_aio_read(s->image, off, size, buf, c); > } > > return &acb->common; > } > > -static BlockDriverAIOCB *rbd_aio_readv(BlockDriverState * bs, > - int64_t sector_num, QEMUIOVector * > qiov, > - int nb_sectors, > - BlockDriverCompletionFunc * cb, > - void *opaque) > +static BlockDriverAIOCB *qemu_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) > +static BlockDriverAIOCB *qemu_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) > +static int qemu_rbd_getinfo(BlockDriverState *bs, BlockDriverInfo *bdi) > { > BDRVRBDState *s = bs->opaque; > - bdi->cluster_size = s->objsize; > + rbd_image_info_t info; > + int r; > + > + r = rbd_stat(s->image, &info, sizeof(info)); > + if (r < 0) { > + return r; > + } > + > + bdi->cluster_size = info.obj_size; > return 0; > } > > -static int64_t rbd_getlength(BlockDriverState * bs) > +static int64_t qemu_rbd_getlength(BlockDriverState *bs) > { > BDRVRBDState *s = bs->opaque; > + rbd_image_info_t info; > + int r; > > - return s->size; > + r = rbd_stat(s->image, &info, sizeof(info)); > + if (r < 0) { > + return r; > + } > + > + return info.size; > } > > -static int rbd_snap_create(BlockDriverState *bs, QEMUSnapshotInfo *sn_info) > +static int qemu_rbd_snap_create(BlockDriverState *bs, > + QEMUSnapshotInfo *sn_info) > { > BDRVRBDState *s = bs->opaque; > - char inbuf[512], outbuf[128]; > - uint64_t snap_id; > int r; > - char *p = inbuf; > - char *end = inbuf + sizeof(inbuf); > - char n[RBD_MAX_SEG_NAME_SIZE]; > - char *hbuf = NULL; > - RbdHeader1 *header; > > if (sn_info->name[0] == '\0') { > return -EINVAL; /* we need a name for rbd snapshots */ > @@ -841,185 +609,59 @@ static int rbd_snap_create(BlockDriverState *bs, > QEMUSnapshotInfo *sn_info) > return -ERANGE; > } > > - r = rados_selfmanaged_snap_create(s->header_pool, &snap_id); > + r = rbd_snap_create(s->image, sn_info->name); > if (r < 0) { > - error_report("failed to create snap id: %s", strerror(-r)); > + error_report("failed to create snap: %s", strerror(-r)); > return r; > } > > - *(uint32_t *)p = strlen(sn_info->name); > - cpu_to_le32s((uint32_t *)p); > - p += sizeof(uint32_t); > - strncpy(p, sn_info->name, end - p); > - p += strlen(p); > - if (p + sizeof(snap_id) > end) { > - error_report("invalid input parameter"); > - return -EINVAL; > - } > - > - *(uint64_t *)p = snap_id; > - cpu_to_le64s((uint64_t *)p); > - > - snprintf(n, sizeof(n), "%s%s", s->name, RBD_SUFFIX); > - > - r = rados_exec(s->header_pool, n, "rbd", "snap_add", inbuf, > - sizeof(inbuf), outbuf, sizeof(outbuf)); > - if (r < 0) { > - error_report("rbd.snap_add execution failed failed: %s", > strerror(-r)); > - return r; > - } > - > - sprintf(sn_info->id_str, "%s", sn_info->name); > - > - r = rbd_read_header(s, &hbuf); > - if (r < 0) { > - error_report("failed reading header: %s", strerror(-r)); > - return r; > - } > - > - header = (RbdHeader1 *) hbuf; > - r = rbd_set_snapc(s->pool, sn_info->name, header); > - if (r < 0) { > - error_report("failed setting snap context: %s", strerror(-r)); > - goto failed; > - } > - > - return 0; > - > -failed: > - qemu_free(header); > - return r; > -} > - > -static int decode32(char **p, const char *end, uint32_t *v) > -{ > - if (*p + 4 > end) { > - return -ERANGE; > - } > - > - *v = *(uint32_t *)(*p); > - le32_to_cpus(v); > - *p += 4; > return 0; > } > > -static int decode64(char **p, const char *end, uint64_t *v) > -{ > - if (*p + 8 > end) { > - return -ERANGE; > - } > - > - *v = *(uint64_t *)(*p); > - le64_to_cpus(v); > - *p += 8; > - return 0; > -} > - > -static int decode_str(char **p, const char *end, char **s) > -{ > - uint32_t len; > - int r; > - > - if ((r = decode32(p, end, &len)) < 0) { > - return r; > - } > - > - *s = qemu_malloc(len + 1); > - memcpy(*s, *p, len); > - *p += len; > - (*s)[len] = '\0'; > - > - return len; > -} > - > -static int rbd_snap_list(BlockDriverState *bs, QEMUSnapshotInfo **psn_tab) > +static int qemu_rbd_snap_list(BlockDriverState *bs, > + QEMUSnapshotInfo **psn_tab) > { > BDRVRBDState *s = bs->opaque; > - char n[RBD_MAX_SEG_NAME_SIZE]; > QEMUSnapshotInfo *sn_info, *sn_tab = NULL; > - RbdHeader1 *header; > - char *hbuf = NULL; > - char *outbuf = NULL, *end, *buf; > - uint64_t len; > - uint64_t snap_seq; > - uint32_t snap_count; > int r, i; > + rbd_snap_info_t *snaps; > + int max_snaps = 100, snap_count; I think it would be nicer to have this defined at the beginning (e.g. RBD_MAX_SNAPS). > > - /* read header to estimate how much space we need to read the snap > - * list */ > - if ((r = rbd_read_header(s, &hbuf)) < 0) { > - goto done_err; > - } > - header = (RbdHeader1 *)hbuf; > - len = le64_to_cpu(header->snap_names_len); > - len += 1024; /* should have already been enough, but new snapshots might > - already been created since we read the header. just > allocate > - a bit more, so that in most cases it'll suffice anyway */ > - qemu_free(hbuf); > - > - snprintf(n, sizeof(n), "%s%s", s->name, RBD_SUFFIX); > - while (1) { > - qemu_free(outbuf); > - outbuf = qemu_malloc(len); > - > - r = rados_exec(s->header_pool, n, "rbd", "snap_list", NULL, 0, > - outbuf, len); > + do { > + snaps = qemu_malloc(sizeof(*snaps) * max_snaps); > + r = rbd_snap_list(s->image, snaps, &max_snaps); > if (r < 0) { > - error_report("rbd.snap_list execution failed failed: %s", > strerror(-r)); > - goto done_err; > + qemu_free(snaps); > } > - if (r != len) { > - break; > - } > + } while (r == -ERANGE); > > - /* if we're here, we probably raced with some snaps creation */ > - len *= 2; > + if (r <= 0) { > + return r; > } > - buf = outbuf; > - end = buf + len; > > - if ((r = decode64(&buf, end, &snap_seq)) < 0) { > - goto done_err; > - } > - if ((r = decode32(&buf, end, &snap_count)) < 0) { > - goto done_err; > - } > + snap_count = r; This isn't really necessary. We could just use snap_count above (or r below). > > sn_tab = qemu_mallocz(snap_count * sizeof(QEMUSnapshotInfo)); > - for (i = 0; i < snap_count; i++) { > - uint64_t id, image_size; > - char *snap_name; > > - if ((r = decode64(&buf, end, &id)) < 0) { > - goto done_err; > - } > - if ((r = decode64(&buf, end, &image_size)) < 0) { > - goto done_err; > - } > - if ((r = decode_str(&buf, end, &snap_name)) < 0) { > - goto done_err; > - } > + for (i = 0; i < snap_count; i++) { > + const char *snap_name = snaps[i].name; > > sn_info = sn_tab + i; > pstrcpy(sn_info->id_str, sizeof(sn_info->id_str), snap_name); > pstrcpy(sn_info->name, sizeof(sn_info->name), snap_name); > - qemu_free(snap_name); > > - sn_info->vm_state_size = image_size; > + sn_info->vm_state_size = snaps[i].size; > sn_info->date_sec = 0; > sn_info->date_nsec = 0; > sn_info->vm_clock_nsec = 0; > } > + rbd_snap_list_end(snaps); > + > *psn_tab = sn_tab; > - qemu_free(outbuf); > return snap_count; > -done_err: > - qemu_free(sn_tab); > - qemu_free(outbuf); > - return r; > } > > -static QEMUOptionParameter rbd_create_options[] = { > +static QEMUOptionParameter qemu_rbd_create_options[] = { > { > .name = BLOCK_OPT_SIZE, > .type = OPT_SIZE, > @@ -1036,19 +678,19 @@ static QEMUOptionParameter rbd_create_options[] = { > static BlockDriver bdrv_rbd = { > .format_name = "rbd", > .instance_size = sizeof(BDRVRBDState), > - .bdrv_file_open = rbd_open, > - .bdrv_close = rbd_close, > - .bdrv_create = rbd_create, > - .bdrv_get_info = rbd_getinfo, > - .create_options = rbd_create_options, > - .bdrv_getlength = rbd_getlength, > + .bdrv_file_open = qemu_rbd_open, > + .bdrv_close = qemu_rbd_close, > + .bdrv_create = qemu_rbd_create, > + .bdrv_get_info = qemu_rbd_getinfo, > + .create_options = qemu_rbd_create_options, > + .bdrv_getlength = qemu_rbd_getlength, > .protocol_name = "rbd", > > - .bdrv_aio_readv = rbd_aio_readv, > - .bdrv_aio_writev = rbd_aio_writev, > + .bdrv_aio_readv = qemu_rbd_aio_readv, > + .bdrv_aio_writev = qemu_rbd_aio_writev, > > - .bdrv_snapshot_create = rbd_snap_create, > - .bdrv_snapshot_list = rbd_snap_list, > + .bdrv_snapshot_create = qemu_rbd_snap_create, > + .bdrv_snapshot_list = qemu_rbd_snap_list, > }; > > static void bdrv_rbd_init(void) > diff --git a/block/rbd_types.h b/block/rbd_types.h > deleted file mode 100644 > index f4cca99..0000000 > --- a/block/rbd_types.h > +++ /dev/null > @@ -1,71 +0,0 @@ > -/* > - * Ceph - scalable distributed file system > - * > - * Copyright (C) 2004-2010 Sage Weil <s...@newdream.net> > - * > - * This is free software; you can redistribute it and/or > - * modify it under the terms of the GNU Lesser General Public > - * License version 2.1, as published by the Free Software > - * Foundation. See file COPYING.LIB. > - * > - */ > - > -#ifndef CEPH_RBD_TYPES_H > -#define CEPH_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_INFO "rbd_info" > - > -#define RBD_DEFAULT_OBJ_ORDER 22 /* 4MB */ > - > -#define RBD_MAX_OBJ_NAME_SIZE 96 > -#define RBD_MAX_BLOCK_NAME_SIZE 24 > -#define RBD_MAX_SEG_NAME_SIZE 128 > - > -#define RBD_COMP_NONE 0 > -#define RBD_CRYPT_NONE 0 > - > -#define RBD_HEADER_TEXT "<<< Rados Block Device Image >>>\n" > -#define RBD_HEADER_SIGNATURE "RBD" > -#define RBD_HEADER_VERSION "001.005" > - > -struct rbd_info { > - uint64_t max_id; > -} __attribute__ ((packed)); > - > -struct rbd_obj_snap_ondisk { > - uint64_t id; > - uint64_t image_size; > -} __attribute__((packed)); > - > -struct rbd_obj_header_ondisk { > - char text[40]; > - char block_name[RBD_MAX_BLOCK_NAME_SIZE]; > - 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 a318d37..378c238 100755 > --- a/configure > +++ b/configure > @@ -1917,41 +1917,24 @@ fi > 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 > - rbd_libs="-lrados" > - if compile_prog "" "$rbd_libs" ; then > - librados_too_old=no > - cat > $TMPC <<EOF > -#include <stdio.h> > -#include <rados/librados.h> > -#ifndef CEPH_OSD_TMAP_SET > -#error missing CEPH_OSD_TMAP_SET > -#endif > +#include <rbd/librbd.h> > int main(void) { > - int (*func)(const rados_pool_t pool, uint64_t *snapid) = > rados_selfmanaged_snap_create; > - rados_initialize(0, NULL); > + rados_t cluster; > + rados_create(&cluster, NULL); > return 0; > } > EOF > - if compile_prog "" "$rbd_libs" ; then > - rbd=yes > - libs_tools="$rbd_libs $libs_tools" > - libs_softmmu="$rbd_libs $libs_softmmu" > - else > - rbd=no > - librados_too_old=yes > - fi > + rbd_libs="-lrbd -lrados" > + if compile_prog "" "$rbd_libs" ; then > + rbd=yes > + libs_tools="$rbd_libs $libs_tools" > + libs_softmmu="$rbd_libs $libs_softmmu" > else > if test "$rbd" = "yes" ; then > feature_not_found "rados block device" > fi > rbd=no > fi > - if test "$librados_too_old" = "yes" ; then > - echo "-> Your librados version is too old - upgrade needed to have rbd > support" > - fi > fi > > ########################################## > -- > 1.7.2.3 > > > -- To unsubscribe from this list: send the line "unsubscribe ceph-devel" in the body of a message to majord...@vger.kernel.org More majordomo info at http://vger.kernel.org/majordomo-info.html