#include <inttypes.h>
#include "qemu-common.h"
-#include "qemu-error.h"
-#include "block_int.h"
+#include "qemu/error-report.h"
+#include "block/block_int.h"
#include <rbd/librbd.h>
typedef enum {
RBD_AIO_READ,
RBD_AIO_WRITE,
- RBD_AIO_DISCARD
+ RBD_AIO_DISCARD,
+ RBD_AIO_FLUSH
} RBDAIOCmd;
typedef struct RBDAIOCB {
BlockDriverAIOCB common;
QEMUBH *bh;
- int ret;
+ int64_t ret;
QEMUIOVector *qiov;
char *bounce;
RBDAIOCmd cmd;
int error;
struct BDRVRBDState *s;
int cancelled;
+ int status;
} RBDAIOCB;
typedef struct RADOSCB {
int done;
int64_t size;
char *buf;
- int ret;
+ int64_t ret;
} RADOSCB;
#define RBD_FD_READ 0
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;
return ret;
}
-static int qemu_rbd_create(const char *filename, QEMUOptionParameter *options)
+static int qemu_rbd_create(const char *filename, QEMUOptionParameter *options,
+ Error **errp)
{
int64_t bytes = 0;
int64_t objsize;
RBDAIOCB *acb = rcb->acb;
int64_t r;
- if (acb->cancelled) {
- qemu_vfree(acb->bounce);
- qemu_aio_release(acb);
- goto done;
- }
-
r = rcb->ret;
- if (acb->cmd == RBD_AIO_WRITE ||
- acb->cmd == RBD_AIO_DISCARD) {
+ if (acb->cmd != RBD_AIO_READ) {
if (r < 0) {
acb->ret = r;
acb->error = 1;
/* Note that acb->bh can be NULL in case where the aio was cancelled */
acb->bh = qemu_bh_new(rbd_aio_bh_cb, acb);
qemu_bh_schedule(acb->bh);
-done:
g_free(rcb);
}
if (s->event_reader_pos == sizeof(s->event_rcb)) {
s->event_reader_pos = 0;
qemu_rbd_complete_aio(s->event_rcb);
- s->qemu_aio_count--;
}
}
} while (ret < 0 && errno == EINTR);
}
-static int qemu_rbd_aio_flush_cb(void *opaque)
-{
- BDRVRBDState *s = opaque;
-
- return (s->qemu_aio_count > 0);
-}
+/* TODO Convert to fine grained options */
+static QemuOptsList runtime_opts = {
+ .name = "rbd",
+ .head = QTAILQ_HEAD_INITIALIZER(runtime_opts.head),
+ .desc = {
+ {
+ .name = "filename",
+ .type = QEMU_OPT_STRING,
+ .help = "Specification of the rbd image",
+ },
+ { /* end of list */ }
+ },
+};
-static int qemu_rbd_open(BlockDriverState *bs, const char *filename, int flags)
+static int qemu_rbd_open(BlockDriverState *bs, QDict *options, int flags,
+ Error **errp)
{
BDRVRBDState *s = bs->opaque;
char pool[RBD_MAX_POOL_NAME_SIZE];
char conf[RBD_MAX_CONF_SIZE];
char clientname_buf[RBD_MAX_CONF_SIZE];
char *clientname;
+ QemuOpts *opts;
+ Error *local_err = NULL;
+ const char *filename;
int r;
+ opts = qemu_opts_create_nofail(&runtime_opts);
+ qemu_opts_absorb_qdict(opts, options, &local_err);
+ if (error_is_set(&local_err)) {
+ qerror_report_err(local_err);
+ error_free(local_err);
+ qemu_opts_del(opts);
+ return -EINVAL;
+ }
+
+ filename = qemu_opt_get(opts, "filename");
+
if (qemu_rbd_parsename(filename, pool, sizeof(pool),
snap_buf, sizeof(snap_buf),
s->name, sizeof(s->name),
conf, sizeof(conf)) < 0) {
- return -EINVAL;
+ r = -EINVAL;
+ goto failed_opts;
}
clientname = qemu_rbd_parse_clientname(conf, clientname_buf);
r = rados_create(&s->cluster, clientname);
if (r < 0) {
error_report("error initializing");
- return r;
+ goto failed_opts;
}
s->snap = NULL;
s->snap = g_strdup(snap_buf);
}
+ /*
+ * Fallback to more conservative semantics if setting cache
+ * options fails. Ignore errors from setting rbd_cache because the
+ * only possible error is that the option does not exist, and
+ * librbd defaults to no caching. If write through caching cannot
+ * be set up, fall back to no caching.
+ */
+ if (flags & BDRV_O_NOCACHE) {
+ rados_conf_set(s->cluster, "rbd_cache", "false");
+ } else {
+ rados_conf_set(s->cluster, "rbd_cache", "true");
+ }
+
if (strstr(conf, "conf=") == NULL) {
/* try default location, but ignore failure */
rados_conf_read_file(s->cluster, NULL);
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], qemu_rbd_aio_event_reader,
- NULL, qemu_rbd_aio_flush_cb, s);
+ NULL, s);
+ qemu_opts_del(opts);
return 0;
failed:
failed_shutdown:
rados_shutdown(s->cluster);
g_free(s->snap);
+failed_opts:
+ qemu_opts_del(opts);
return r;
}
close(s->fds[0]);
close(s->fds[1]);
- qemu_aio_set_fd_handler(s->fds[RBD_FD_READ], NULL, NULL, NULL, NULL);
+ qemu_aio_set_fd_handler(s->fds[RBD_FD_READ], NULL, NULL, NULL);
rbd_close(s->image);
rados_ioctx_destroy(s->io_ctx);
{
RBDAIOCB *acb = (RBDAIOCB *) blockacb;
acb->cancelled = 1;
+
+ while (acb->status == -EINPROGRESS) {
+ qemu_aio_wait();
+ }
+
+ qemu_aio_release(acb);
}
-static AIOPool rbd_aio_pool = {
+static const AIOCBInfo rbd_aiocb_info = {
.aiocb_size = sizeof(RBDAIOCB),
.cancel = qemu_rbd_aio_cancel,
};
acb->common.cb(acb->common.opaque, (acb->ret > 0 ? 0 : acb->ret));
qemu_bh_delete(acb->bh);
acb->bh = NULL;
+ acb->status = 0;
- qemu_aio_release(acb);
+ if (!acb->cancelled) {
+ qemu_aio_release(acb);
+ }
}
static int rbd_aio_discard_wrapper(rbd_image_t image,
#endif
}
+static int rbd_aio_flush_wrapper(rbd_image_t image,
+ rbd_completion_t comp)
+{
+#ifdef LIBRBD_SUPPORTS_AIO_FLUSH
+ return rbd_aio_flush(image, comp);
+#else
+ return -ENOTSUP;
+#endif
+}
+
static BlockDriverAIOCB *rbd_start_aio(BlockDriverState *bs,
int64_t sector_num,
QEMUIOVector *qiov,
BDRVRBDState *s = bs->opaque;
- acb = qemu_aio_get(&rbd_aio_pool, bs, cb, opaque);
+ acb = qemu_aio_get(&rbd_aiocb_info, bs, cb, opaque);
acb->cmd = cmd;
acb->qiov = qiov;
- if (cmd == RBD_AIO_DISCARD) {
+ if (cmd == RBD_AIO_DISCARD || cmd == RBD_AIO_FLUSH) {
acb->bounce = NULL;
} else {
acb->bounce = qemu_blockalign(bs, qiov->size);
acb->s = s;
acb->cancelled = 0;
acb->bh = NULL;
+ acb->status = -EINPROGRESS;
if (cmd == RBD_AIO_WRITE) {
- qemu_iovec_to_buffer(acb->qiov, acb->bounce);
+ qemu_iovec_to_buf(acb->qiov, 0, acb->bounce, qiov->size);
}
buf = acb->bounce;
off = sector_num * BDRV_SECTOR_SIZE;
size = nb_sectors * BDRV_SECTOR_SIZE;
- s->qemu_aio_count++; /* All the RADOSCB */
-
rcb = g_malloc(sizeof(RADOSCB));
rcb->done = 0;
rcb->acb = acb;
case RBD_AIO_DISCARD:
r = rbd_aio_discard_wrapper(s->image, off, size, c);
break;
+ case RBD_AIO_FLUSH:
+ r = rbd_aio_flush_wrapper(s->image, c);
+ break;
default:
r = -EINVAL;
}
failed:
g_free(rcb);
- s->qemu_aio_count--;
qemu_aio_release(acb);
return NULL;
}
RBD_AIO_WRITE);
}
+#ifdef LIBRBD_SUPPORTS_AIO_FLUSH
+static BlockDriverAIOCB *qemu_rbd_aio_flush(BlockDriverState *bs,
+ BlockDriverCompletionFunc *cb,
+ void *opaque)
+{
+ return rbd_start_aio(bs, 0, NULL, 0, cb, opaque, RBD_AIO_FLUSH);
+}
+
+#else
+
static int qemu_rbd_co_flush(BlockDriverState *bs)
{
#if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 1)
return 0;
#endif
}
+#endif
static int qemu_rbd_getinfo(BlockDriverState *bs, BlockDriverInfo *bdi)
{
}
static int qemu_rbd_snap_remove(BlockDriverState *bs,
- const char *snapshot_name)
+ const char *snapshot_id,
+ const char *snapshot_name,
+ Error **errp)
{
BDRVRBDState *s = bs->opaque;
int r;
+ if (!snapshot_name) {
+ error_setg(errp, "rbd need a valid snapshot name");
+ return -EINVAL;
+ }
+
+ /* If snapshot_id is specified, it must be equal to name, see
+ qemu_rbd_snap_list() */
+ if (snapshot_id && strcmp(snapshot_id, snapshot_name)) {
+ error_setg(errp,
+ "rbd do not support snapshot id, it should be NULL or "
+ "equal to snapshot name");
+ return -EINVAL;
+ }
+
r = rbd_snap_remove(s->image, snapshot_name);
+ if (r < 0) {
+ error_setg_errno(errp, -r, "Failed to remove the snapshot");
+ }
return r;
}
static BlockDriver bdrv_rbd = {
.format_name = "rbd",
.instance_size = sizeof(BDRVRBDState),
+ .bdrv_needs_filename = true,
.bdrv_file_open = qemu_rbd_open,
.bdrv_close = qemu_rbd_close,
.bdrv_create = qemu_rbd_create,
+ .bdrv_has_zero_init = bdrv_has_zero_init_1,
.bdrv_get_info = qemu_rbd_getinfo,
.create_options = qemu_rbd_create_options,
.bdrv_getlength = qemu_rbd_getlength,
.bdrv_aio_readv = qemu_rbd_aio_readv,
.bdrv_aio_writev = qemu_rbd_aio_writev,
+
+#ifdef LIBRBD_SUPPORTS_AIO_FLUSH
+ .bdrv_aio_flush = qemu_rbd_aio_flush,
+#else
.bdrv_co_flush_to_disk = qemu_rbd_co_flush,
+#endif
#ifdef LIBRBD_SUPPORTS_DISCARD
.bdrv_aio_discard = qemu_rbd_aio_discard,
projects / qemu.git / blobdiff