fix live migration

[qemu.git] / buffered_file.c

diff --git a/buffered_file.c b/buffered_file.c
index 59d952df0ad92eace831737de93187a78145dbb6..bd0f61d8c9684ca8347be8dfdc9d0619219613a5 100644 (file)
--- a/buffered_file.c
+++ b/buffered_file.c
@@ -23,10 +23,6 @@
 
 typedef struct QEMUFileBuffered
 {
-    BufferedPutFunc *put_buffer;
-    BufferedPutReadyFunc *put_ready;
-    BufferedWaitForUnfreezeFunc *wait_for_unfreeze;
-    BufferedCloseFunc *close;
     MigrationState *migration_state;
     QEMUFile *file;
     int freeze_output;
@@ -62,33 +58,26 @@ static void buffered_append(QEMUFileBuffered *s,
     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);
@@ -100,12 +89,17 @@ static void buffered_flush(QEMUFileBuffered *s)
     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);
 
@@ -123,13 +117,17 @@ static int buffered_put_buffer(void *opaque, const uint8_t *buf, int64_t pos, in
         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);
         }
     }
 
@@ -139,19 +137,29 @@ static int buffered_put_buffer(void *opaque, const uint8_t *buf, int64_t pos, in
 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);
@@ -166,6 +174,13 @@ static int buffered_close(void *opaque)
  *   1: Time to stop
  *   negative: There has been an error
  */
+static int buffered_get_fd(void *opaque)
+{
+    QEMUFileBuffered *s = opaque;
+
+    return qemu_get_fd(s->file);
+}
+
 static int buffered_rate_limit(void *opaque)
 {
     QEMUFileBuffered *s = opaque;
@@ -226,28 +241,25 @@ static void buffered_rate_tick(void *opaque)
     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)
+static const QEMUFileOps buffered_file_ops = {
+    .get_fd =         buffered_get_fd,
+    .put_buffer =     buffered_put_buffer,
+    .close =          buffered_close,
+    .rate_limit =     buffered_rate_limit,
+    .get_rate_limit = buffered_get_rate_limit,
+    .set_rate_limit = buffered_set_rate_limit,
+};
+
+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->file = qemu_fopen_ops(s, buffered_put_buffer, NULL,
-                             buffered_close, buffered_rate_limit,
-                             buffered_set_rate_limit,
-                            buffered_get_rate_limit);
+    s->xfer_limit = migration_state->bandwidth_limit / 10;
+
+    s->file = qemu_fopen_ops(s, &buffered_file_ops);
 
     s->timer = qemu_new_timer_ms(rt_clock, buffered_rate_tick, s);