typedef struct QEMUFileBuffered
{
- BufferedPutFunc *put_buffer;
- BufferedPutReadyFunc *put_ready;
- BufferedWaitForUnfreezeFunc *wait_for_unfreeze;
- BufferedCloseFunc *close;
MigrationState *migration_state;
QEMUFile *file;
int freeze_output;
s->buffer_size += size;
}
-static void buffered_flush(QEMUFileBuffered *s)
+static ssize_t buffered_flush(QEMUFileBuffered *s)
{
size_t offset = 0;
- int error;
-
- error = qemu_file_get_error(s->file);
- if (error != 0) {
- DPRINTF("flush when error, bailing: %s\n", strerror(-error));
- return;
- }
+ ssize_t ret = 0;
DPRINTF("flushing %zu byte(s) of data\n", s->buffer_size);
while (s->bytes_xfer < s->xfer_limit && offset < s->buffer_size) {
- ssize_t ret;
- ret = s->put_buffer(s->migration_state, 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);
- qemu_file_set_error(s->file, ret);
break;
} else {
DPRINTF("flushed %zd byte(s)\n", 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 error;
+ ssize_t error;
DPRINTF("putting %d bytes at %" PRId64 "\n", size, pos);
buffered_append(s, buf, size);
}
- buffered_flush(s);
+ 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->freeze_output && s->bytes_xfer < s->xfer_limit) {
DPRINTF("notifying client\n");
- s->put_ready(s->migration_state);
+ migrate_fd_put_ready(s->migration_state);
}
}
static int buffered_close(void *opaque)
{
QEMUFileBuffered *s = opaque;
- int ret;
+ ssize_t ret = 0;
+ int ret2;
DPRINTF("closing\n");
s->xfer_limit = INT_MAX;
while (!qemu_file_get_error(s->file) && s->buffer_size) {
- buffered_flush(s);
- if (s->freeze_output)
- s->wait_for_unfreeze(s->migration_state);
+ 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->migration_state);
-
+ ret2 = migrate_fd_close(s->migration_state);
+ if (ret >= 0) {
+ ret = ret2;
+ }
qemu_del_timer(s->timer);
qemu_free_timer(s->timer);
g_free(s->buffer);
buffered_put_buffer(s, NULL, 0, 0);
}
-QEMUFile *qemu_fopen_ops_buffered(MigrationState *migration_state,
- 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 = g_malloc0(sizeof(*s));
s->migration_state = migration_state;
- 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->xfer_limit = migration_state->bandwidth_limit / 10;
s->file = qemu_fopen_ops(s, buffered_put_buffer, NULL,
buffered_close, buffered_rate_limit,