* This work is licensed under the terms of the GNU GPL, version 2. See
* the COPYING file in the top-level directory.
*
+ * Contributions after 2012-01-13 are licensed under the terms of the
+ * GNU GPL, version 2 or (at your option) any later version.
*/
#include "qemu-common.h"
#include "hw/hw.h"
#include "qemu-timer.h"
-#include "sysemu.h"
#include "qemu-char.h"
#include "buffered_file.h"
typedef struct QEMUFileBuffered
{
- BufferedPutFunc *put_buffer;
- BufferedPutReadyFunc *put_ready;
- BufferedWaitForUnfreezeFunc *wait_for_unfreeze;
- BufferedCloseFunc *close;
- void *opaque;
+ MigrationState *migration_state;
QEMUFile *file;
- int has_error;
int freeze_output;
size_t bytes_xfer;
size_t xfer_limit;
const uint8_t *buf, size_t size)
{
if (size > (s->buffer_capacity - s->buffer_size)) {
- void *tmp;
-
DPRINTF("increasing buffer capacity from %zu by %zu\n",
s->buffer_capacity, size + 1024);
s->buffer_capacity += size + 1024;
- tmp = qemu_realloc(s->buffer, s->buffer_capacity);
- if (tmp == NULL) {
- fprintf(stderr, "qemu file buffer expansion failed\n");
- exit(1);
- }
-
- s->buffer = tmp;
+ s->buffer = g_realloc(s->buffer, s->buffer_capacity);
}
memcpy(s->buffer + s->buffer_size, buf, size);
s->buffer_size += size;
}
-static void buffered_flush(QEMUFileBuffered *s)
+static ssize_t buffered_flush(QEMUFileBuffered *s)
{
size_t offset = 0;
-
- if (s->has_error) {
- DPRINTF("flush when error, bailing\n");
- return;
- }
+ ssize_t ret = 0;
DPRINTF("flushing %zu byte(s) of data\n", s->buffer_size);
- while (offset < s->buffer_size) {
- ssize_t ret;
+ while (s->bytes_xfer < s->xfer_limit && offset < s->buffer_size) {
- ret = s->put_buffer(s->opaque, s->buffer + offset,
- s->buffer_size - offset);
+ ret = migrate_fd_put_buffer(s->migration_state, s->buffer + offset,
+ s->buffer_size - offset);
if (ret == -EAGAIN) {
DPRINTF("backend not ready, freezing\n");
+ ret = 0;
s->freeze_output = 1;
break;
}
if (ret <= 0) {
DPRINTF("error flushing data, %zd\n", ret);
- s->has_error = 1;
break;
} else {
DPRINTF("flushed %zd byte(s)\n", ret);
offset += ret;
+ s->bytes_xfer += ret;
}
}
DPRINTF("flushed %zu of %zu byte(s)\n", offset, s->buffer_size);
memmove(s->buffer, s->buffer + offset, s->buffer_size - offset);
s->buffer_size -= offset;
+
+ if (ret < 0) {
+ return ret;
+ }
+ return offset;
}
static int buffered_put_buffer(void *opaque, const uint8_t *buf, int64_t pos, int size)
{
QEMUFileBuffered *s = opaque;
- int offset = 0;
- ssize_t ret;
+ ssize_t error;
DPRINTF("putting %d bytes at %" PRId64 "\n", size, pos);
- if (s->has_error) {
- DPRINTF("flush when error, bailing\n");
- return -EINVAL;
+ error = qemu_file_get_error(s->file);
+ if (error) {
+ DPRINTF("flush when error, bailing: %s\n", strerror(-error));
+ return error;
}
DPRINTF("unfreezing output\n");
s->freeze_output = 0;
- buffered_flush(s);
-
- while (!s->freeze_output && offset < size) {
- if (s->bytes_xfer > s->xfer_limit) {
- DPRINTF("transfer limit exceeded when putting\n");
- break;
- }
-
- ret = s->put_buffer(s->opaque, buf + offset, size - offset);
- if (ret == -EAGAIN) {
- DPRINTF("backend not ready, freezing\n");
- s->freeze_output = 1;
- break;
- }
-
- if (ret <= 0) {
- DPRINTF("error putting\n");
- s->has_error = 1;
- offset = -EINVAL;
- break;
- }
-
- DPRINTF("put %zd byte(s)\n", ret);
- offset += ret;
- s->bytes_xfer += ret;
+ if (size > 0) {
+ DPRINTF("buffering %d bytes\n", size - offset);
+ buffered_append(s, buf, size);
}
- if (offset >= 0) {
- DPRINTF("buffering %d bytes\n", size - offset);
- buffered_append(s, buf + offset, size - offset);
- offset = size;
+ error = buffered_flush(s);
+ if (error < 0) {
+ DPRINTF("buffered flush error. bailing: %s\n", strerror(-error));
+ return error;
}
if (pos == 0 && size == 0) {
DPRINTF("file is ready\n");
- if (s->bytes_xfer <= s->xfer_limit) {
+ if (!s->freeze_output && s->bytes_xfer < s->xfer_limit) {
DPRINTF("notifying client\n");
- s->put_ready(s->opaque);
+ migrate_fd_put_ready(s->migration_state);
}
}
- return offset;
+ return size;
}
static int buffered_close(void *opaque)
{
QEMUFileBuffered *s = opaque;
- int ret;
+ ssize_t ret = 0;
+ int ret2;
DPRINTF("closing\n");
- while (!s->has_error && s->buffer_size) {
- buffered_flush(s);
- if (s->freeze_output)
- s->wait_for_unfreeze(s);
+ s->xfer_limit = INT_MAX;
+ while (!qemu_file_get_error(s->file) && s->buffer_size) {
+ ret = buffered_flush(s);
+ if (ret < 0) {
+ break;
+ }
+ if (s->freeze_output) {
+ ret = migrate_fd_wait_for_unfreeze(s->migration_state);
+ if (ret < 0) {
+ break;
+ }
+ }
}
- ret = s->close(s->opaque);
-
+ ret2 = migrate_fd_close(s->migration_state);
+ if (ret >= 0) {
+ ret = ret2;
+ }
qemu_del_timer(s->timer);
qemu_free_timer(s->timer);
- qemu_free(s->buffer);
- qemu_free(s);
+ g_free(s->buffer);
+ g_free(s);
return ret;
}
+/*
+ * The meaning of the return values is:
+ * 0: We can continue sending
+ * 1: Time to stop
+ * negative: There has been an error
+ */
static int buffered_rate_limit(void *opaque)
{
QEMUFileBuffered *s = opaque;
+ int ret;
- if (s->has_error)
- return 0;
-
+ ret = qemu_file_get_error(s->file);
+ if (ret) {
+ return ret;
+ }
if (s->freeze_output)
return 1;
static int64_t buffered_set_rate_limit(void *opaque, int64_t new_rate)
{
QEMUFileBuffered *s = opaque;
- if (s->has_error)
+ if (qemu_file_get_error(s->file)) {
goto out;
-
+ }
if (new_rate > SIZE_MAX) {
new_rate = SIZE_MAX;
}
{
QEMUFileBuffered *s = opaque;
- if (s->has_error) {
+ if (qemu_file_get_error(s->file)) {
buffered_close(s);
return;
}
s->bytes_xfer = 0;
- buffered_flush(s);
-
- /* Add some checks around this */
- s->put_ready(s->opaque);
+ buffered_put_buffer(s, NULL, 0, 0);
}
-QEMUFile *qemu_fopen_ops_buffered(void *opaque,
- size_t bytes_per_sec,
- BufferedPutFunc *put_buffer,
- BufferedPutReadyFunc *put_ready,
- BufferedWaitForUnfreezeFunc *wait_for_unfreeze,
- BufferedCloseFunc *close)
+QEMUFile *qemu_fopen_ops_buffered(MigrationState *migration_state)
{
QEMUFileBuffered *s;
- s = qemu_mallocz(sizeof(*s));
+ s = g_malloc0(sizeof(*s));
- s->opaque = opaque;
- s->xfer_limit = bytes_per_sec / 10;
- s->put_buffer = put_buffer;
- s->put_ready = put_ready;
- s->wait_for_unfreeze = wait_for_unfreeze;
- s->close = close;
+ s->migration_state = migration_state;
+ s->xfer_limit = migration_state->bandwidth_limit / 10;
s->file = qemu_fopen_ops(s, buffered_put_buffer, NULL,
buffered_close, buffered_rate_limit,
projects / qemu.git / blobdiff