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 void buffered_flush(QEMUFileBuffered *s)
  62 {
  63     size_t offset = 0;
  64     int error;
  65
  66     error = qemu_file_get_error(s->file);
  67     if (error != 0) {
  68         DPRINTF("flush when error, bailing: %s\n", strerror(-error));
  69         return;
  70     }
  71
  72     DPRINTF("flushing %zu byte(s) of data\n", s->buffer_size);
  73
  74     while (s->bytes_xfer < s->xfer_limit && offset < s->buffer_size) {
  75         ssize_t ret;
  76
  77         ret = migrate_fd_put_buffer(s->migration_state, s->buffer + offset,
  78                                     s->buffer_size - offset);
  79         if (ret == -EAGAIN) {
  80             DPRINTF("backend not ready, freezing\n");
  81             s->freeze_output = 1;
  82             break;
  83         }
  84
  85         if (ret <= 0) {
  86             DPRINTF("error flushing data, %zd\n", ret);
  87             qemu_file_set_error(s->file, ret);
  88             break;
  89         } else {
  90             DPRINTF("flushed %zd byte(s)\n", ret);
  91             offset += ret;
  92             s->bytes_xfer += ret;
  93         }
  94     }
  95
  96     DPRINTF("flushed %zu of %zu byte(s)\n", offset, s->buffer_size);
  97     memmove(s->buffer, s->buffer + offset, s->buffer_size - offset);
  98     s->buffer_size -= offset;
  99 }
 100
 101 static int buffered_put_buffer(void *opaque, const uint8_t *buf, int64_t pos, int size)
 102 {
 103     QEMUFileBuffered *s = opaque;
 104     int error;
 105
 106     DPRINTF("putting %d bytes at %" PRId64 "\n", size, pos);
 107
 108     error = qemu_file_get_error(s->file);
 109     if (error) {
 110         DPRINTF("flush when error, bailing: %s\n", strerror(-error));
 111         return error;
 112     }
 113
 114     DPRINTF("unfreezing output\n");
 115     s->freeze_output = 0;
 116
 117     if (size > 0) {
 118         DPRINTF("buffering %d bytes\n", size - offset);
 119         buffered_append(s, buf, size);
 120     }
 121
 122     buffered_flush(s);
 123
 124     if (pos == 0 && size == 0) {
 125         DPRINTF("file is ready\n");
 126         if (!s->freeze_output && s->bytes_xfer < s->xfer_limit) {
 127             DPRINTF("notifying client\n");
 128             migrate_fd_put_ready(s->migration_state);
 129         }
 130     }
 131
 132     return size;
 133 }
 134
 135 static int buffered_close(void *opaque)
 136 {
 137     QEMUFileBuffered *s = opaque;
 138     int ret;
 139
 140     DPRINTF("closing\n");
 141
 142     s->xfer_limit = INT_MAX;
 143     while (!qemu_file_get_error(s->file) && s->buffer_size) {
 144         buffered_flush(s);
 145         if (s->freeze_output)
 146             migrate_fd_wait_for_unfreeze(s->migration_state);
 147     }
 148
 149     ret = migrate_fd_close(s->migration_state);
 150
 151     qemu_del_timer(s->timer);
 152     qemu_free_timer(s->timer);
 153     g_free(s->buffer);
 154     g_free(s);
 155
 156     return ret;
 157 }
 158
 159 /*
 160  * The meaning of the return values is:
 161  *   0: We can continue sending
 162  *   1: Time to stop
 163  *   negative: There has been an error
 164  */
 165 static int buffered_rate_limit(void *opaque)
 166 {
 167     QEMUFileBuffered *s = opaque;
 168     int ret;
 169
 170     ret = qemu_file_get_error(s->file);
 171     if (ret) {
 172         return ret;
 173     }
 174     if (s->freeze_output)
 175         return 1;
 176
 177     if (s->bytes_xfer > s->xfer_limit)
 178         return 1;
 179
 180     return 0;
 181 }
 182
 183 static int64_t buffered_set_rate_limit(void *opaque, int64_t new_rate)
 184 {
 185     QEMUFileBuffered *s = opaque;
 186     if (qemu_file_get_error(s->file)) {
 187         goto out;
 188     }
 189     if (new_rate > SIZE_MAX) {
 190         new_rate = SIZE_MAX;
 191     }
 192
 193     s->xfer_limit = new_rate / 10;
 194
 195 out:
 196     return s->xfer_limit;
 197 }
 198
 199 static int64_t buffered_get_rate_limit(void *opaque)
 200 {
 201     QEMUFileBuffered *s = opaque;
 202
 203     return s->xfer_limit;
 204 }
 205
 206 static void buffered_rate_tick(void *opaque)
 207 {
 208     QEMUFileBuffered *s = opaque;
 209
 210     if (qemu_file_get_error(s->file)) {
 211         buffered_close(s);
 212         return;
 213     }
 214
 215     qemu_mod_timer(s->timer, qemu_get_clock_ms(rt_clock) + 100);
 216
 217     if (s->freeze_output)
 218         return;
 219
 220     s->bytes_xfer = 0;
 221
 222     buffered_put_buffer(s, NULL, 0, 0);
 223 }
 224
 225 QEMUFile *qemu_fopen_ops_buffered(MigrationState *migration_state,
 226                                   size_t bytes_per_sec)
 227 {
 228     QEMUFileBuffered *s;
 229
 230     s = g_malloc0(sizeof(*s));
 231
 232     s->migration_state = migration_state;
 233     s->xfer_limit = bytes_per_sec / 10;
 234
 235     s->file = qemu_fopen_ops(s, buffered_put_buffer, NULL,
 236                              buffered_close, buffered_rate_limit,
 237                              buffered_set_rate_limit,
 238                              buffered_get_rate_limit);
 239
 240     s->timer = qemu_new_timer_ms(rt_clock, buffered_rate_tick, s);
 241
 242     qemu_mod_timer(s->timer, qemu_get_clock_ms(rt_clock) + 100);
 243
 244     return s->file;
 245 }