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