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