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         qemu_file_set_error(s->file, error);
 124
 125         return error;
 126     }
 127
 128     if (pos == 0 && size == 0) {
 129         DPRINTF("file is ready\n");
 130         if (!s->freeze_output && s->bytes_xfer < s->xfer_limit) {
 131             DPRINTF("notifying client\n");
 132             migrate_fd_put_ready(s->migration_state);
 133         }
 134     }
 135
 136     return size;
 137 }
 138
 139 static int buffered_close(void *opaque)
 140 {
 141     QEMUFileBuffered *s = opaque;
 142     ssize_t ret = 0;
 143     int ret2;
 144
 145     DPRINTF("closing\n");
 146
 147     s->xfer_limit = INT_MAX;
 148     while (!qemu_file_get_error(s->file) && s->buffer_size) {
 149         ret = buffered_flush(s);
 150         if (ret < 0) {
 151             break;
 152         }
 153         if (s->freeze_output) {
 154             ret = migrate_fd_wait_for_unfreeze(s->migration_state);
 155             if (ret < 0) {
 156                 break;
 157             }
 158         }
 159     }
 160
 161     ret2 = migrate_fd_close(s->migration_state);
 162     if (ret >= 0) {
 163         ret = ret2;
 164     }
 165     qemu_del_timer(s->timer);
 166     qemu_free_timer(s->timer);
 167     g_free(s->buffer);
 168     g_free(s);
 169
 170     return ret;
 171 }
 172
 173 /*
 174  * The meaning of the return values is:
 175  *   0: We can continue sending
 176  *   1: Time to stop
 177  *   negative: There has been an error
 178  */
 179 static int buffered_rate_limit(void *opaque)
 180 {
 181     QEMUFileBuffered *s = opaque;
 182     int ret;
 183
 184     ret = qemu_file_get_error(s->file);
 185     if (ret) {
 186         return ret;
 187     }
 188     if (s->freeze_output)
 189         return 1;
 190
 191     if (s->bytes_xfer > s->xfer_limit)
 192         return 1;
 193
 194     return 0;
 195 }
 196
 197 static int64_t buffered_set_rate_limit(void *opaque, int64_t new_rate)
 198 {
 199     QEMUFileBuffered *s = opaque;
 200     if (qemu_file_get_error(s->file)) {
 201         goto out;
 202     }
 203     if (new_rate > SIZE_MAX) {
 204         new_rate = SIZE_MAX;
 205     }
 206
 207     s->xfer_limit = new_rate / 10;
 208
 209 out:
 210     return s->xfer_limit;
 211 }
 212
 213 static int64_t buffered_get_rate_limit(void *opaque)
 214 {
 215     QEMUFileBuffered *s = opaque;
 216
 217     return s->xfer_limit;
 218 }
 219
 220 static void buffered_rate_tick(void *opaque)
 221 {
 222     QEMUFileBuffered *s = opaque;
 223
 224     if (qemu_file_get_error(s->file)) {
 225         buffered_close(s);
 226         return;
 227     }
 228
 229     qemu_mod_timer(s->timer, qemu_get_clock_ms(rt_clock) + 100);
 230
 231     if (s->freeze_output)
 232         return;
 233
 234     s->bytes_xfer = 0;
 235
 236     buffered_put_buffer(s, NULL, 0, 0);
 237 }
 238
 239 QEMUFile *qemu_fopen_ops_buffered(MigrationState *migration_state)
 240 {
 241     QEMUFileBuffered *s;
 242
 243     s = g_malloc0(sizeof(*s));
 244
 245     s->migration_state = migration_state;
 246     s->xfer_limit = migration_state->bandwidth_limit / 10;
 247
 248     s->file = qemu_fopen_ops(s, buffered_put_buffer, NULL,
 249                              buffered_close, buffered_rate_limit,
 250                              buffered_set_rate_limit,
 251                              buffered_get_rate_limit);
 252
 253     s->timer = qemu_new_timer_ms(rt_clock, buffered_rate_tick, s);
 254
 255     qemu_mod_timer(s->timer, qemu_get_clock_ms(rt_clock) + 100);
 256
 257     return s->file;
 258 }