#define MAX_IOV_SIZE MIN_CONST(IOV_MAX, 64)
struct QEMUFile {
- const QEMUFileOps *ops;
const QEMUFileHooks *hooks;
- void *opaque;
+ QIOChannel *ioc;
+ bool is_writable;
- int64_t bytes_xfer;
- int64_t xfer_limit;
+ /*
+ * Maximum amount of data in bytes to transfer during one
+ * rate limiting time window
+ */
+ int64_t rate_limit_max;
+ /*
+ * Total amount of data in bytes queued for transfer
+ * during this rate limiting time window
+ */
+ int64_t rate_limit_used;
+
+ /* The sum of bytes transferred on the wire */
+ int64_t total_transferred;
- int64_t pos; /* start of buffer when writing, end of buffer
- when reading */
int buf_index;
int buf_size; /* 0 when writing */
uint8_t buf[IO_BUF_SIZE];
Error *last_error_obj;
/* has the file has been shutdown */
bool shutdown;
- /* Whether opaque points to a QIOChannel */
- bool has_ioc;
};
/*
* Stop a file from being read/written - not all backing files can do this
* typically only sockets can.
+ *
+ * TODO: convert to propagate Error objects instead of squashing
+ * to a fixed errno value
*/
int qemu_file_shutdown(QEMUFile *f)
{
- int ret;
+ int ret = 0;
f->shutdown = true;
- if (!f->ops->shut_down) {
+ if (!qio_channel_has_feature(f->ioc,
+ QIO_CHANNEL_FEATURE_SHUTDOWN)) {
return -ENOSYS;
}
- ret = f->ops->shut_down(f->opaque, true, true, NULL);
+
+ if (qio_channel_shutdown(f->ioc, QIO_CHANNEL_SHUTDOWN_BOTH, NULL) < 0) {
+ ret = -EIO;
+ }
if (!f->last_error) {
qemu_file_set_error(f, -EIO);
return ret;
}
-/*
- * Result: QEMUFile* for a 'return path' for comms in the opposite direction
- * NULL if not available
- */
-QEMUFile *qemu_file_get_return_path(QEMUFile *f)
-{
- if (!f->ops->get_return_path) {
- return NULL;
- }
- return f->ops->get_return_path(f->opaque);
-}
-
bool qemu_file_mode_is_not_valid(const char *mode)
{
if (mode == NULL ||
return false;
}
-QEMUFile *qemu_fopen_ops(void *opaque, const QEMUFileOps *ops, bool has_ioc)
+static QEMUFile *qemu_file_new_impl(QIOChannel *ioc, bool is_writable)
{
QEMUFile *f;
f = g_new0(QEMUFile, 1);
- f->opaque = opaque;
- f->ops = ops;
- f->has_ioc = has_ioc;
+ object_ref(ioc);
+ f->ioc = ioc;
+ f->is_writable = is_writable;
+
return f;
}
+/*
+ * Result: QEMUFile* for a 'return path' for comms in the opposite direction
+ * NULL if not available
+ */
+QEMUFile *qemu_file_get_return_path(QEMUFile *f)
+{
+ return qemu_file_new_impl(f->ioc, !f->is_writable);
+}
+
+QEMUFile *qemu_file_new_output(QIOChannel *ioc)
+{
+ return qemu_file_new_impl(ioc, true);
+}
+
+QEMUFile *qemu_file_new_input(QIOChannel *ioc)
+{
+ return qemu_file_new_impl(ioc, false);
+}
void qemu_file_set_hooks(QEMUFile *f, const QEMUFileHooks *hooks)
{
return f->last_error;
}
+/*
+ * Get last error for either stream f1 or f2 with optional Error*.
+ * The error returned (non-zero) can be either from f1 or f2.
+ *
+ * If any of the qemufile* is NULL, then skip the check on that file.
+ *
+ * When there is no error on both qemufile, zero is returned.
+ */
+int qemu_file_get_error_obj_any(QEMUFile *f1, QEMUFile *f2, Error **errp)
+{
+ int ret = 0;
+
+ if (f1) {
+ ret = qemu_file_get_error_obj(f1, errp);
+ /* If there's already error detected, return */
+ if (ret) {
+ return ret;
+ }
+ }
+
+ if (f2) {
+ ret = qemu_file_get_error_obj(f2, errp);
+ }
+
+ return ret;
+}
+
/*
* Set the last error for stream f with optional Error*
*/
bool qemu_file_is_writable(QEMUFile *f)
{
- return f->ops->writev_buffer;
+ return f->is_writable;
}
static void qemu_iovec_release_ram(QEMUFile *f)
memset(f->may_free, 0, sizeof(f->may_free));
}
+
/**
* Flushes QEMUFile buffer
*
*/
void qemu_fflush(QEMUFile *f)
{
- ssize_t ret = 0;
- ssize_t expect = 0;
- Error *local_error = NULL;
-
if (!qemu_file_is_writable(f)) {
return;
}
return;
}
if (f->iovcnt > 0) {
- expect = iov_size(f->iov, f->iovcnt);
- ret = f->ops->writev_buffer(f->opaque, f->iov, f->iovcnt, f->pos,
- &local_error);
+ Error *local_error = NULL;
+ if (qio_channel_writev_all(f->ioc,
+ f->iov, f->iovcnt,
+ &local_error) < 0) {
+ qemu_file_set_error_obj(f, -EIO, local_error);
+ } else {
+ f->total_transferred += iov_size(f->iov, f->iovcnt);
+ }
qemu_iovec_release_ram(f);
}
- if (ret >= 0) {
- f->pos += ret;
- }
- /* We expect the QEMUFile write impl to send the full
- * data set we requested, so sanity check that.
- */
- if (ret != expect) {
- qemu_file_set_error_obj(f, ret < 0 ? ret : -EIO, local_error);
- }
f->buf_index = 0;
f->iovcnt = 0;
}
int ret = 0;
if (f->hooks && f->hooks->before_ram_iterate) {
- ret = f->hooks->before_ram_iterate(f, f->opaque, flags, NULL);
+ ret = f->hooks->before_ram_iterate(f, flags, NULL);
if (ret < 0) {
qemu_file_set_error(f, ret);
}
int ret = 0;
if (f->hooks && f->hooks->after_ram_iterate) {
- ret = f->hooks->after_ram_iterate(f, f->opaque, flags, NULL);
+ ret = f->hooks->after_ram_iterate(f, flags, NULL);
if (ret < 0) {
qemu_file_set_error(f, ret);
}
int ret = -EINVAL;
if (f->hooks && f->hooks->hook_ram_load) {
- ret = f->hooks->hook_ram_load(f, f->opaque, flags, data);
+ ret = f->hooks->hook_ram_load(f, flags, data);
if (ret < 0) {
qemu_file_set_error(f, ret);
}
uint64_t *bytes_sent)
{
if (f->hooks && f->hooks->save_page) {
- int ret = f->hooks->save_page(f, f->opaque, block_offset,
+ int ret = f->hooks->save_page(f, block_offset,
offset, size, bytes_sent);
if (ret != RAM_SAVE_CONTROL_NOT_SUPP) {
- f->bytes_xfer += size;
+ f->rate_limit_used += size;
}
if (ret != RAM_SAVE_CONTROL_DELAYED &&
ret != RAM_SAVE_CONTROL_NOT_SUPP) {
if (bytes_sent && *bytes_sent > 0) {
- qemu_update_position(f, *bytes_sent);
+ qemu_file_credit_transfer(f, *bytes_sent);
} else if (ret < 0) {
qemu_file_set_error(f, ret);
}
return 0;
}
- len = f->ops->get_buffer(f->opaque, f->buf + pending, f->pos,
- IO_BUF_SIZE - pending, &local_error);
+ do {
+ len = qio_channel_read(f->ioc,
+ (char *)f->buf + pending,
+ IO_BUF_SIZE - pending,
+ &local_error);
+ if (len == QIO_CHANNEL_ERR_BLOCK) {
+ if (qemu_in_coroutine()) {
+ qio_channel_yield(f->ioc, G_IO_IN);
+ } else {
+ qio_channel_wait(f->ioc, G_IO_IN);
+ }
+ } else if (len < 0) {
+ len = -EIO;
+ }
+ } while (len == QIO_CHANNEL_ERR_BLOCK);
+
if (len > 0) {
f->buf_size += len;
- f->pos += len;
+ f->total_transferred += len;
} else if (len == 0) {
qemu_file_set_error_obj(f, -EIO, local_error);
} else if (len != -EAGAIN) {
return len;
}
-void qemu_update_position(QEMUFile *f, size_t size)
+void qemu_file_credit_transfer(QEMUFile *f, size_t size)
{
- f->pos += size;
+ f->total_transferred += size;
}
/** Closes the file
*/
int qemu_fclose(QEMUFile *f)
{
- int ret;
+ int ret, ret2;
qemu_fflush(f);
ret = qemu_file_get_error(f);
- if (f->ops->close) {
- int ret2 = f->ops->close(f->opaque, NULL);
- if (ret >= 0) {
- ret = ret2;
- }
+ ret2 = qio_channel_close(f->ioc, NULL);
+ if (ret >= 0) {
+ ret = ret2;
}
+ g_clear_pointer(&f->ioc, object_unref);
+
/* If any error was spotted before closing, we should report it
* instead of the close() return value.
*/
return;
}
- f->bytes_xfer += size;
+ f->rate_limit_used += size;
add_to_iovec(f, buf, size, may_free);
}
l = size;
}
memcpy(f->buf + f->buf_index, buf, l);
- f->bytes_xfer += l;
+ f->rate_limit_used += l;
add_buf_to_iovec(f, l);
if (qemu_file_get_error(f)) {
break;
}
f->buf[f->buf_index] = v;
- f->bytes_xfer++;
+ f->rate_limit_used++;
add_buf_to_iovec(f, 1);
}
return result;
}
-int64_t qemu_ftell_fast(QEMUFile *f)
+int64_t qemu_file_total_transferred_fast(QEMUFile *f)
{
- int64_t ret = f->pos;
+ int64_t ret = f->total_transferred;
int i;
for (i = 0; i < f->iovcnt; i++) {
return ret;
}
-int64_t qemu_ftell(QEMUFile *f)
+int64_t qemu_file_total_transferred(QEMUFile *f)
{
qemu_fflush(f);
- return f->pos;
+ return f->total_transferred;
}
int qemu_file_rate_limit(QEMUFile *f)
if (qemu_file_get_error(f)) {
return 1;
}
- if (f->xfer_limit > 0 && f->bytes_xfer > f->xfer_limit) {
+ if (f->rate_limit_max > 0 && f->rate_limit_used > f->rate_limit_max) {
return 1;
}
return 0;
int64_t qemu_file_get_rate_limit(QEMUFile *f)
{
- return f->xfer_limit;
+ return f->rate_limit_max;
}
void qemu_file_set_rate_limit(QEMUFile *f, int64_t limit)
{
- f->xfer_limit = limit;
+ f->rate_limit_max = limit;
}
void qemu_file_reset_rate_limit(QEMUFile *f)
{
- f->bytes_xfer = 0;
+ f->rate_limit_used = 0;
}
-void qemu_file_update_transfer(QEMUFile *f, int64_t len)
+void qemu_file_acct_rate_limit(QEMUFile *f, int64_t len)
{
- f->bytes_xfer += len;
+ f->rate_limit_used += len;
}
void qemu_put_be16(QEMUFile *f, unsigned int v)
*/
void qemu_file_set_blocking(QEMUFile *f, bool block)
{
- if (f->ops->set_blocking) {
- f->ops->set_blocking(f->opaque, block, NULL);
- }
+ qio_channel_set_blocking(f->ioc, block, NULL);
}
/*
- * Return the ioc object if it's a migration channel. Note: it can return NULL
- * for callers passing in a non-migration qemufile. E.g. see qemu_fopen_bdrv()
- * and its usage in e.g. load_snapshot(). So we need to check against NULL
- * before using it. If without the check, migration_incoming_state_destroy()
- * could fail for load_snapshot().
+ * qemu_file_get_ioc:
+ *
+ * Get the ioc object for the file, without incrementing
+ * the reference count.
+ *
+ * Returns: the ioc object
*/
QIOChannel *qemu_file_get_ioc(QEMUFile *file)
{
- return file->has_ioc ? QIO_CHANNEL(file->opaque) : NULL;
+ return file->ioc;
}
projects / qemu.git / blobdiff