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     BufferedPutFunc *put_buffer;
  27     BufferedPutReadyFunc *put_ready;
  28     BufferedWaitForUnfreezeFunc *wait_for_unfreeze;
  29     BufferedCloseFunc *close;
  30     MigrationState *migration_state;
  31     QEMUFile *file;
  32     int freeze_output;
  33     size_t bytes_xfer;
  34     size_t xfer_limit;
  35     uint8_t *buffer;
  36     size_t buffer_size;
  37     size_t buffer_capacity;
  38     QEMUTimer *timer;
  39 } QEMUFileBuffered;
  40
  41 #ifdef DEBUG_BUFFERED_FILE
  42 #define DPRINTF(fmt, ...) \
  43     do { printf("buffered-file: " fmt, ## __VA_ARGS__); } while (0)
  44 #else
  45 #define DPRINTF(fmt, ...) \
  46     do { } while (0)
  47 #endif
  48
  49 static void buffered_append(QEMUFileBuffered *s,
  50                             const uint8_t *buf, size_t size)
  51 {
  52     if (size > (s->buffer_capacity - s->buffer_size)) {
  53         DPRINTF("increasing buffer capacity from %zu by %zu\n",
  54                 s->buffer_capacity, size + 1024);
  55
  56         s->buffer_capacity += size + 1024;
  57
  58         s->buffer = g_realloc(s->buffer, s->buffer_capacity);
  59     }
  60
  61     memcpy(s->buffer + s->buffer_size, buf, size);
  62     s->buffer_size += size;
  63 }
  64
  65 static void buffered_flush(QEMUFileBuffered *s)
  66 {
  67     size_t offset = 0;
  68     int error;
  69
  70     error = qemu_file_get_error(s->file);
  71     if (error != 0) {
  72         DPRINTF("flush when error, bailing: %s\n", strerror(-error));
  73         return;
  74     }
  75
  76     DPRINTF("flushing %zu byte(s) of data\n", s->buffer_size);
  77
  78     while (s->bytes_xfer < s->xfer_limit && offset < s->buffer_size) {
  79         ssize_t ret;
  80
  81         ret = s->put_buffer(s->migration_state, s->buffer + offset,
  82                             s->buffer_size - offset);
  83         if (ret == -EAGAIN) {
  84             DPRINTF("backend not ready, freezing\n");
  85             s->freeze_output = 1;
  86             break;
  87         }
  88
  89         if (ret <= 0) {
  90             DPRINTF("error flushing data, %zd\n", ret);
  91             qemu_file_set_error(s->file, ret);
  92             break;
  93         } else {
  94             DPRINTF("flushed %zd byte(s)\n", ret);
  95             offset += ret;
  96             s->bytes_xfer += ret;
  97         }
  98     }
  99
 100     DPRINTF("flushed %zu of %zu byte(s)\n", offset, s->buffer_size);
 101     memmove(s->buffer, s->buffer + offset, s->buffer_size - offset);
 102     s->buffer_size -= offset;
 103 }
 104
 105 static int buffered_put_buffer(void *opaque, const uint8_t *buf, int64_t pos, int size)
 106 {
 107     QEMUFileBuffered *s = opaque;
 108     int error;
 109
 110     DPRINTF("putting %d bytes at %" PRId64 "\n", size, pos);
 111
 112     error = qemu_file_get_error(s->file);
 113     if (error) {
 114         DPRINTF("flush when error, bailing: %s\n", strerror(-error));
 115         return error;
 116     }
 117
 118     DPRINTF("unfreezing output\n");
 119     s->freeze_output = 0;
 120
 121     if (size > 0) {
 122         DPRINTF("buffering %d bytes\n", size - offset);
 123         buffered_append(s, buf, size);
 124     }
 125
 126     buffered_flush(s);
 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             s->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     int ret;
 143
 144     DPRINTF("closing\n");
 145
 146     s->xfer_limit = INT_MAX;
 147     while (!qemu_file_get_error(s->file) && s->buffer_size) {
 148         buffered_flush(s);
 149         if (s->freeze_output)
 150             s->wait_for_unfreeze(s->migration_state);
 151     }
 152
 153     ret = s->close(s->migration_state);
 154
 155     qemu_del_timer(s->timer);
 156     qemu_free_timer(s->timer);
 157     g_free(s->buffer);
 158     g_free(s);
 159
 160     return ret;
 161 }
 162
 163 /*
 164  * The meaning of the return values is:
 165  *   0: We can continue sending
 166  *   1: Time to stop
 167  *   negative: There has been an error
 168  */
 169 static int buffered_rate_limit(void *opaque)
 170 {
 171     QEMUFileBuffered *s = opaque;
 172     int ret;
 173
 174     ret = qemu_file_get_error(s->file);
 175     if (ret) {
 176         return ret;
 177     }
 178     if (s->freeze_output)
 179         return 1;
 180
 181     if (s->bytes_xfer > s->xfer_limit)
 182         return 1;
 183
 184     return 0;
 185 }
 186
 187 static int64_t buffered_set_rate_limit(void *opaque, int64_t new_rate)
 188 {
 189     QEMUFileBuffered *s = opaque;
 190     if (qemu_file_get_error(s->file)) {
 191         goto out;
 192     }
 193     if (new_rate > SIZE_MAX) {
 194         new_rate = SIZE_MAX;
 195     }
 196
 197     s->xfer_limit = new_rate / 10;
 198
 199 out:
 200     return s->xfer_limit;
 201 }
 202
 203 static int64_t buffered_get_rate_limit(void *opaque)
 204 {
 205     QEMUFileBuffered *s = opaque;
 206
 207     return s->xfer_limit;
 208 }
 209
 210 static void buffered_rate_tick(void *opaque)
 211 {
 212     QEMUFileBuffered *s = opaque;
 213
 214     if (qemu_file_get_error(s->file)) {
 215         buffered_close(s);
 216         return;
 217     }
 218
 219     qemu_mod_timer(s->timer, qemu_get_clock_ms(rt_clock) + 100);
 220
 221     if (s->freeze_output)
 222         return;
 223
 224     s->bytes_xfer = 0;
 225
 226     buffered_put_buffer(s, NULL, 0, 0);
 227 }
 228
 229 QEMUFile *qemu_fopen_ops_buffered(MigrationState *migration_state,
 230                                   size_t bytes_per_sec,
 231                                   BufferedPutFunc *put_buffer,
 232                                   BufferedPutReadyFunc *put_ready,
 233                                   BufferedWaitForUnfreezeFunc *wait_for_unfreeze,
 234                                   BufferedCloseFunc *close)
 235 {
 236     QEMUFileBuffered *s;
 237
 238     s = g_malloc0(sizeof(*s));
 239
 240     s->migration_state = migration_state;
 241     s->xfer_limit = bytes_per_sec / 10;
 242     s->put_buffer = put_buffer;
 243     s->put_ready = put_ready;
 244     s->wait_for_unfreeze = wait_for_unfreeze;
 245     s->close = close;
 246
 247     s->file = qemu_fopen_ops(s, buffered_put_buffer, NULL,
 248                              buffered_close, buffered_rate_limit,
 249                              buffered_set_rate_limit,
 250                              buffered_get_rate_limit);
 251
 252     s->timer = qemu_new_timer_ms(rt_clock, buffered_rate_tick, s);
 253
 254     qemu_mod_timer(s->timer, qemu_get_clock_ms(rt_clock) + 100);
 255
 256     return s->file;
 257 }