buffered_file.c

   1 /*
   2  * QEMU buffered QEMUFile
   3  *
   4  * Copyright IBM, Corp. 2008
   5  *
   6  * Authors:
   7  *  Anthony Liguori   <aliguori@us.ibm.com>
   8  *
   9  * This work is licensed under the terms of the GNU GPL, version 2.  See
  10  * the COPYING file in the top-level directory.
  11  *
  12  * Contributions after 2012-01-13 are licensed under the terms of the
  13  * GNU GPL, version 2 or (at your option) any later version.
  14  */
  15
  16 #include "qemu-common.h"
  17 #include "hw/hw.h"
  18 #include "qemu-timer.h"
  19 #include "qemu-char.h"
  20 #include "buffered_file.h"
  21
  22 //#define DEBUG_BUFFERED_FILE
  23
  24 typedef struct QEMUFileBuffered
  25 {
  26     MigrationState *migration_state;
  27     QEMUFile *file;
  28     int freeze_output;
  29     size_t bytes_xfer;
  30     size_t xfer_limit;
  31     uint8_t *buffer;
  32     size_t buffer_size;
  33     size_t buffer_capacity;
  34     QEMUTimer *timer;
  35 } QEMUFileBuffered;
  36
  37 #ifdef DEBUG_BUFFERED_FILE
  38 #define DPRINTF(fmt, ...) \
  39     do { printf("buffered-file: " fmt, ## __VA_ARGS__); } while (0)
  40 #else
  41 #define DPRINTF(fmt, ...) \
  42     do { } while (0)
  43 #endif
  44
  45 static void buffered_append(QEMUFileBuffered *s,
  46                             const uint8_t *buf, size_t size)
  47 {
  48     if (size > (s->buffer_capacity - s->buffer_size)) {
  49         DPRINTF("increasing buffer capacity from %zu by %zu\n",
  50                 s->buffer_capacity, size + 1024);
  51
  52         s->buffer_capacity += size + 1024;
  53
  54         s->buffer = g_realloc(s->buffer, s->buffer_capacity);
  55     }
  56
  57     memcpy(s->buffer + s->buffer_size, buf, size);
  58     s->buffer_size += size;
  59 }
  60
  61 static ssize_t buffered_flush(QEMUFileBuffered *s)
  62 {
  63     size_t offset = 0;
  64     ssize_t ret = 0;
  65
  66     DPRINTF("flushing %zu byte(s) of data\n", s->buffer_size);
  67
  68     while (s->bytes_xfer < s->xfer_limit && offset < s->buffer_size) {
  69
  70         ret = migrate_fd_put_buffer(s->migration_state, s->buffer + offset,
  71                                     s->buffer_size - offset);
  72         if (ret == -EAGAIN) {
  73             DPRINTF("backend not ready, freezing\n");
  74             ret = 0;
  75             s->freeze_output = 1;
  76             break;
  77         }
  78
  79         if (ret <= 0) {
  80             DPRINTF("error flushing data, %zd\n", ret);
  81             break;
  82         } else {
  83             DPRINTF("flushed %zd byte(s)\n", ret);
  84             offset += ret;
  85             s->bytes_xfer += ret;
  86         }
  87     }
  88
  89     DPRINTF("flushed %zu of %zu byte(s)\n", offset, s->buffer_size);
  90     memmove(s->buffer, s->buffer + offset, s->buffer_size - offset);
  91     s->buffer_size -= offset;
  92
  93     if (ret < 0) {
  94         return ret;
  95     }
  96     return offset;
  97 }
  98
  99 static int buffered_put_buffer(void *opaque, const uint8_t *buf, int64_t pos, int size)
 100 {
 101     QEMUFileBuffered *s = opaque;
 102     ssize_t error;
 103
 104     DPRINTF("putting %d bytes at %" PRId64 "\n", size, pos);
 105
 106     error = qemu_file_get_error(s->file);
 107     if (error) {
 108         DPRINTF("flush when error, bailing: %s\n", strerror(-error));
 109         return error;
 110     }
 111
 112     DPRINTF("unfreezing output\n");
 113     s->freeze_output = 0;
 114
 115     if (size > 0) {
 116         DPRINTF("buffering %d bytes\n", size - offset);
 117         buffered_append(s, buf, size);
 118     }
 119
 120     error = buffered_flush(s);
 121     if (error < 0) {
 122         DPRINTF("buffered flush error. bailing: %s\n", strerror(-error));
 123         return error;
 124     }
 125
 126     if (pos == 0 && size == 0) {
 127         DPRINTF("file is ready\n");
 128         if (!s->freeze_output && s->bytes_xfer < s->xfer_limit) {
 129             DPRINTF("notifying client\n");
 130             migrate_fd_put_ready(s->migration_state);
 131         }
 132     }
 133
 134     return size;
 135 }
 136
 137 static int buffered_close(void *opaque)
 138 {
 139     QEMUFileBuffered *s = opaque;
 140     ssize_t ret = 0;
 141     int ret2;
 142
 143     DPRINTF("closing\n");
 144
 145     s->xfer_limit = INT_MAX;
 146     while (!qemu_file_get_error(s->file) && s->buffer_size) {
 147         ret = buffered_flush(s);
 148         if (ret < 0) {
 149             break;
 150         }
 151         if (s->freeze_output) {
 152             ret = migrate_fd_wait_for_unfreeze(s->migration_state);
 153             if (ret < 0) {
 154                 break;
 155             }
 156         }
 157     }
 158
 159     ret2 = migrate_fd_close(s->migration_state);
 160     if (ret >= 0) {
 161         ret = ret2;
 162     }
 163     qemu_del_timer(s->timer);
 164     qemu_free_timer(s->timer);
 165     g_free(s->buffer);
 166     g_free(s);
 167
 168     return ret;
 169 }
 170
 171 /*
 172  * The meaning of the return values is:
 173  *   0: We can continue sending
 174  *   1: Time to stop
 175  *   negative: There has been an error
 176  */
 177 static int buffered_rate_limit(void *opaque)
 178 {
 179     QEMUFileBuffered *s = opaque;
 180     int ret;
 181
 182     ret = qemu_file_get_error(s->file);
 183     if (ret) {
 184         return ret;
 185     }
 186     if (s->freeze_output)
 187         return 1;
 188
 189     if (s->bytes_xfer > s->xfer_limit)
 190         return 1;
 191
 192     return 0;
 193 }
 194
 195 static int64_t buffered_set_rate_limit(void *opaque, int64_t new_rate)
 196 {
 197     QEMUFileBuffered *s = opaque;
 198     if (qemu_file_get_error(s->file)) {
 199         goto out;
 200     }
 201     if (new_rate > SIZE_MAX) {
 202         new_rate = SIZE_MAX;
 203     }
 204
 205     s->xfer_limit = new_rate / 10;
 206
 207 out:
 208     return s->xfer_limit;
 209 }
 210
 211 static int64_t buffered_get_rate_limit(void *opaque)
 212 {
 213     QEMUFileBuffered *s = opaque;
 214
 215     return s->xfer_limit;
 216 }
 217
 218 static void buffered_rate_tick(void *opaque)
 219 {
 220     QEMUFileBuffered *s = opaque;
 221
 222     if (qemu_file_get_error(s->file)) {
 223         buffered_close(s);
 224         return;
 225     }
 226
 227     qemu_mod_timer(s->timer, qemu_get_clock_ms(rt_clock) + 100);
 228
 229     if (s->freeze_output)
 230         return;
 231
 232     s->bytes_xfer = 0;
 233
 234     buffered_put_buffer(s, NULL, 0, 0);
 235 }
 236
 237 QEMUFile *qemu_fopen_ops_buffered(MigrationState *migration_state)
 238 {
 239     QEMUFileBuffered *s;
 240
 241     s = g_malloc0(sizeof(*s));
 242
 243     s->migration_state = migration_state;
 244     s->xfer_limit = migration_state->bandwidth_limit / 10;
 245
 246     s->file = qemu_fopen_ops(s, buffered_put_buffer, NULL,
 247                              buffered_close, buffered_rate_limit,
 248                              buffered_set_rate_limit,
 249                              buffered_get_rate_limit);
 250
 251     s->timer = qemu_new_timer_ms(rt_clock, buffered_rate_tick, s);
 252
 253     qemu_mod_timer(s->timer, qemu_get_clock_ms(rt_clock) + 100);
 254
 255     return s->file;
 256 }