migration-tcp.c

   1 /*
   2  * QEMU live migration
   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  */
  13
  14 #include "qemu-common.h"
  15 #include "qemu_socket.h"
  16 #include "migration.h"
  17 #include "qemu-char.h"
  18 #include "sysemu.h"
  19 #include "console.h"
  20 #include "buffered_file.h"
  21 #include "block.h"
  22
  23 //#define DEBUG_MIGRATION_TCP
  24
  25 typedef struct FdMigrationState
  26 {
  27     MigrationState mig_state;
  28     QEMUFile *file;
  29     int64_t bandwidth_limit;
  30     int fd;
  31     int detach;
  32     int state;
  33 } FdMigrationState;
  34
  35 #ifdef DEBUG_MIGRATION_TCP
  36 #define dprintf(fmt, ...) \
  37     do { printf("migration-tcp: " fmt, ## __VA_ARGS__); } while (0)
  38 #else
  39 #define dprintf(fmt, ...) \
  40     do { } while (0)
  41 #endif
  42
  43 static void tcp_cleanup(FdMigrationState *s)
  44 {
  45     qemu_set_fd_handler2(s->fd, NULL, NULL, NULL, NULL);
  46
  47     if (s->file) {
  48         dprintf("closing file\n");
  49         qemu_fclose(s->file);
  50     }
  51
  52     if (s->fd != -1)
  53         close(s->fd);
  54
  55     /* Don't resume monitor until we've flushed all of the buffers */
  56     if (s->detach == 2) {
  57         monitor_resume();
  58         s->detach = 0;
  59     }
  60
  61     s->fd = -1;
  62 }
  63
  64 static void tcp_error(FdMigrationState *s)
  65 {
  66     dprintf("setting error state\n");
  67     s->state = MIG_STATE_ERROR;
  68     tcp_cleanup(s);
  69 }
  70
  71 static void fd_put_notify(void *opaque)
  72 {
  73     FdMigrationState *s = opaque;
  74
  75     qemu_set_fd_handler2(s->fd, NULL, NULL, NULL, NULL);
  76     qemu_file_put_notify(s->file);
  77 }
  78
  79 static ssize_t fd_put_buffer(void *opaque, const void *data, size_t size)
  80 {
  81     FdMigrationState *s = opaque;
  82     ssize_t ret;
  83
  84     do {
  85         ret = send(s->fd, data, size, 0);
  86     } while (ret == -1 && (socket_error() == EINTR || socket_error() == EWOULDBLOCK));
  87
  88     if (ret == -1)
  89         ret = -socket_error();
  90
  91     if (ret == -EAGAIN)
  92         qemu_set_fd_handler2(s->fd, NULL, NULL, fd_put_notify, s);
  93
  94     return ret;
  95 }
  96
  97 static int fd_close(void *opaque)
  98 {
  99     FdMigrationState *s = opaque;
 100     dprintf("fd_close\n");
 101     if (s->fd != -1) {
 102         close(s->fd);
 103         s->fd = -1;
 104     }
 105     return 0;
 106 }
 107
 108 static void fd_wait_for_unfreeze(void *opaque)
 109 {
 110     FdMigrationState *s = opaque;
 111     int ret;
 112
 113     dprintf("wait for unfreeze\n");
 114     if (s->state != MIG_STATE_ACTIVE)
 115         return;
 116
 117     do {
 118         fd_set wfds;
 119
 120         FD_ZERO(&wfds);
 121         FD_SET(s->fd, &wfds);
 122
 123         ret = select(s->fd + 1, NULL, &wfds, NULL, NULL);
 124     } while (ret == -1 && socket_error() == EINTR);
 125 }
 126
 127 static void fd_put_ready(void *opaque)
 128 {
 129     FdMigrationState *s = opaque;
 130
 131     if (s->state != MIG_STATE_ACTIVE) {
 132         dprintf("put_ready returning because of non-active state\n");
 133         return;
 134     }
 135
 136     dprintf("iterate\n");
 137     if (qemu_savevm_state_iterate(s->file) == 1) {
 138         dprintf("done iterating\n");
 139         vm_stop(0);
 140
 141         bdrv_flush_all();
 142         qemu_savevm_state_complete(s->file);
 143         s->state = MIG_STATE_COMPLETED;
 144         tcp_cleanup(s);
 145     }
 146 }
 147
 148 static void tcp_connect_migrate(FdMigrationState *s)
 149 {
 150     int ret;
 151
 152     s->file = qemu_fopen_ops_buffered(s,
 153                                       s->bandwidth_limit,
 154                                       fd_put_buffer,
 155                                       fd_put_ready,
 156                                       fd_wait_for_unfreeze,
 157                                       fd_close);
 158
 159     dprintf("beginning savevm\n");
 160     ret = qemu_savevm_state_begin(s->file);
 161     if (ret < 0) {
 162         dprintf("failed, %d\n", ret);
 163         tcp_error(s);
 164         return;
 165     }
 166
 167     fd_put_ready(s);
 168 }
 169
 170 static void tcp_wait_for_connect(void *opaque)
 171 {
 172     FdMigrationState *s = opaque;
 173     int val, ret;
 174     socklen_t valsize = sizeof(val);
 175
 176     dprintf("connect completed\n");
 177     do {
 178         ret = getsockopt(s->fd, SOL_SOCKET, SO_ERROR, &val, &valsize);
 179     } while (ret == -1 && socket_error() == EINTR);
 180
 181     if (ret < 0) {
 182         tcp_error(s);
 183         return;
 184     }
 185
 186     qemu_set_fd_handler2(s->fd, NULL, NULL, NULL, NULL);
 187
 188     if (val == 0)
 189         tcp_connect_migrate(s);
 190     else {
 191         dprintf("error connecting %d\n", val);
 192         tcp_error(s);
 193     }
 194 }
 195
 196 static FdMigrationState *to_fms(MigrationState *mig_state)
 197 {
 198     return container_of(mig_state, FdMigrationState, mig_state);
 199 }
 200
 201 static int tcp_get_status(MigrationState *mig_state)
 202 {
 203     FdMigrationState *s = to_fms(mig_state);
 204
 205     return s->state;
 206 }
 207
 208 static void tcp_cancel(MigrationState *mig_state)
 209 {
 210     FdMigrationState *s = to_fms(mig_state);
 211
 212     if (s->state != MIG_STATE_ACTIVE)
 213         return;
 214
 215     dprintf("cancelling migration\n");
 216
 217     s->state = MIG_STATE_CANCELLED;
 218
 219     tcp_cleanup(s);
 220 }
 221
 222 static void tcp_release(MigrationState *mig_state)
 223 {
 224     FdMigrationState *s = to_fms(mig_state);
 225
 226     dprintf("releasing state\n");
 227
 228     if (s->state == MIG_STATE_ACTIVE) {
 229         s->state = MIG_STATE_CANCELLED;
 230         tcp_cleanup(s);
 231     }
 232     free(s);
 233 }
 234
 235 MigrationState *tcp_start_outgoing_migration(const char *host_port,
 236                                              int64_t bandwidth_limit,
 237                                              int async)
 238 {
 239     struct sockaddr_in addr;
 240     FdMigrationState *s;
 241     int ret;
 242
 243     if (parse_host_port(&addr, host_port) < 0)
 244         return NULL;
 245
 246     s = qemu_mallocz(sizeof(*s));
 247     if (s == NULL)
 248         return NULL;
 249
 250     s->mig_state.cancel = tcp_cancel;
 251     s->mig_state.get_status = tcp_get_status;
 252     s->mig_state.release = tcp_release;
 253
 254     s->state = MIG_STATE_ACTIVE;
 255     s->detach = !async;
 256     s->bandwidth_limit = bandwidth_limit;
 257     s->fd = socket(PF_INET, SOCK_STREAM, 0);
 258     if (s->fd == -1) {
 259         qemu_free(s);
 260         return NULL;
 261     }
 262
 263     socket_set_nonblock(s->fd);
 264
 265     if (s->detach == 1) {
 266         dprintf("detaching from monitor\n");
 267         monitor_suspend();
 268         s->detach = 2;
 269     }
 270
 271     do {
 272         ret = connect(s->fd, (struct sockaddr *)&addr, sizeof(addr));
 273         if (ret == -1)
 274             ret = -socket_error();
 275
 276         if (ret == -EINPROGRESS || ret == -EWOULDBLOCK)
 277             qemu_set_fd_handler2(s->fd, NULL, NULL, tcp_wait_for_connect, s);
 278     } while (ret == -EINTR);
 279
 280     if (ret < 0 && ret != -EINPROGRESS && ret != -EWOULDBLOCK) {
 281         dprintf("connect failed\n");
 282         close(s->fd);
 283         qemu_free(s);
 284         return NULL;
 285     } else if (ret >= 0)
 286         tcp_connect_migrate(s);
 287
 288     return &s->mig_state;
 289 }
 290
 291 static void tcp_accept_incoming_migration(void *opaque)
 292 {
 293     struct sockaddr_in addr;
 294     socklen_t addrlen = sizeof(addr);
 295     int s = (unsigned long)opaque;
 296     QEMUFile *f;
 297     int c, ret;
 298
 299     do {
 300         c = accept(s, (struct sockaddr *)&addr, &addrlen);
 301     } while (c == -1 && socket_error() == EINTR);
 302
 303     dprintf("accepted migration\n");
 304
 305     if (c == -1) {
 306         fprintf(stderr, "could not accept migration connection\n");
 307         return;
 308     }
 309
 310     f = qemu_fopen_socket(c);
 311     if (f == NULL) {
 312         fprintf(stderr, "could not qemu_fopen socket\n");
 313         goto out;
 314     }
 315
 316     vm_stop(0); /* just in case */
 317     ret = qemu_loadvm_state(f);
 318     if (ret < 0) {
 319         fprintf(stderr, "load of migration failed\n");
 320         goto out_fopen;
 321     }
 322     qemu_announce_self();
 323     dprintf("successfully loaded vm state\n");
 324
 325     /* we've successfully migrated, close the server socket */
 326     qemu_set_fd_handler2(s, NULL, NULL, NULL, NULL);
 327     close(s);
 328
 329     vm_start();
 330
 331 out_fopen:
 332     qemu_fclose(f);
 333 out:
 334     close(c);
 335 }
 336
 337 int tcp_start_incoming_migration(const char *host_port)
 338 {
 339     struct sockaddr_in addr;
 340     int val;
 341     int s;
 342
 343     if (parse_host_port(&addr, host_port) < 0) {
 344         fprintf(stderr, "invalid host/port combination: %s\n", host_port);
 345         return -EINVAL;
 346     }
 347
 348     s = socket(PF_INET, SOCK_STREAM, 0);
 349     if (s == -1)
 350         return -socket_error();
 351
 352     val = 1;
 353     setsockopt(s, SOL_SOCKET, SO_REUSEADDR, (const char *)&val, sizeof(val));
 354
 355     if (bind(s, (struct sockaddr *)&addr, sizeof(addr)) == -1)
 356         goto err;
 357
 358     if (listen(s, 1) == -1)
 359         goto err;
 360
 361     qemu_set_fd_handler2(s, NULL, tcp_accept_incoming_migration, NULL,
 362                          (void *)(unsigned long)s);
 363
 364     return 0;
 365
 366 err:
 367     close(s);
 368     return -socket_error();
 369 }