mirrors/qemu
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'bonzini/migr-coroutine' into staging

Browse Source 
Signed-off-by: Anthony Liguori <aliguori@us.ibm.com>

* bonzini/migr-coroutine:
  migration: move process_incoming_migration to a coroutine
  migration: handle EAGAIN while reading QEMUFile
  migration: move qemu_fclose to process_incoming_migration
  migration: close socket QEMUFile from socket_close
  migration: xxx_close will only be called once
  migration: use closesocket, not close
  migration: use migrate_fd_close in migrate_fd_cleanup
  migration: clean up server sockets and handlers before invoking process_incoming_migration
  migration: replace qemu_stdio_fd with qemu_get_fd
  migration: add qemu_get_fd
  migration: consolidate QEMUFile methods in a single QEMUFileOps struct
  migration: unify stdio-based QEMUFile operations
This commit is contained in:
Anthony Liguori 2012-11-02 13:06:28 -05:00
commit 2a0dfd004d
8 changed files with 215 additions and 154 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
buffered_file.c
View File

@ -174,6 +174,13 @@ static int buffered_close(void *opaque)
* 1: Time to stop
* negative: There has been an error
*/
static int buffered_get_fd(void *opaque)
{
QEMUFileBuffered *s = opaque;
return qemu_get_fd(s->file);
}
static int buffered_rate_limit(void *opaque)
{
QEMUFileBuffered *s = opaque;
@ -234,6 +241,15 @@ static void buffered_rate_tick(void *opaque)
buffered_put_buffer(s, NULL, 0, 0);
}
static const QEMUFileOps buffered_file_ops = {
.get_fd = buffered_get_fd,
.put_buffer = buffered_put_buffer,
.close = buffered_close,
.rate_limit = buffered_rate_limit,
.get_rate_limit = buffered_get_rate_limit,
.set_rate_limit = buffered_set_rate_limit,
};
QEMUFile *qemu_fopen_ops_buffered(MigrationState *migration_state)
{
QEMUFileBuffered *s;
@ -243,10 +259,7 @@ QEMUFile *qemu_fopen_ops_buffered(MigrationState *migration_state)
s->migration_state = migration_state;
s->xfer_limit = migration_state->bandwidth_limit / 10;
s->file = qemu_fopen_ops(s, buffered_put_buffer, NULL,
buffered_close, buffered_rate_limit,
buffered_set_rate_limit,
buffered_get_rate_limit);
s->file = qemu_fopen_ops(s, &buffered_file_ops);
s->timer = qemu_new_timer_ms(rt_clock, buffered_rate_tick, s);

19
migration-exec.c
View File

@ -48,14 +48,12 @@ static int exec_close(MigrationState *s)
{
int ret = 0;
DPRINTF("exec_close\n");
if (s->opaque) {
ret = qemu_fclose(s->opaque);
s->opaque = NULL;
s->fd = -1;
if (ret >= 0 && !(WIFEXITED(ret) && WEXITSTATUS(ret) == 0)) {
/* close succeeded, but non-zero exit code: */
ret = -EIO; /* fake errno value */
}
ret = qemu_fclose(s->opaque);
s->opaque = NULL;
s->fd = -1;
if (ret >= 0 && !(WIFEXITED(ret) && WEXITSTATUS(ret) == 0)) {
/* close succeeded, but non-zero exit code: */
ret = -EIO; /* fake errno value */
}
return ret;
}
@ -87,9 +85,8 @@ static void exec_accept_incoming_migration(void *opaque)
{
QEMUFile *f = opaque;
qemu_set_fd_handler2(qemu_get_fd(f), NULL, NULL, NULL, NULL);
process_incoming_migration(f);
qemu_set_fd_handler2(qemu_stdio_fd(f), NULL, NULL, NULL, NULL);
qemu_fclose(f);
}
void exec_start_incoming_migration(const char *command, Error **errp)
@ -103,6 +100,6 @@ void exec_start_incoming_migration(const char *command, Error **errp)
return;
}
qemu_set_fd_handler2(qemu_stdio_fd(f), NULL,
qemu_set_fd_handler2(qemu_get_fd(f), NULL,
exec_accept_incoming_migration, NULL, f);
}

36
migration-fd.c
View File

@ -48,29 +48,26 @@ static int fd_close(MigrationState *s)
int ret;
DPRINTF("fd_close\n");
if (s->fd != -1) {
ret = fstat(s->fd, &st);
if (ret == 0 && S_ISREG(st.st_mode)) {
/*
* If the file handle is a regular file make sure the
* data is flushed to disk before signaling success.
*/
ret = fsync(s->fd);
if (ret != 0) {
ret = -errno;
perror("migration-fd: fsync");
return ret;
}
}
ret = close(s->fd);
s->fd = -1;
ret = fstat(s->fd, &st);
if (ret == 0 && S_ISREG(st.st_mode)) {
/*
* If the file handle is a regular file make sure the
* data is flushed to disk before signaling success.
*/
ret = fsync(s->fd);
if (ret != 0) {
ret = -errno;
perror("migration-fd: close");
perror("migration-fd: fsync");
return ret;
}
}
return 0;
ret = close(s->fd);
s->fd = -1;
if (ret != 0) {
ret = -errno;
perror("migration-fd: close");
}
return ret;
}
void fd_start_outgoing_migration(MigrationState *s, const char *fdname, Error **errp)
@ -92,9 +89,8 @@ static void fd_accept_incoming_migration(void *opaque)
{
QEMUFile *f = opaque;
qemu_set_fd_handler2(qemu_get_fd(f), NULL, NULL, NULL, NULL);
process_incoming_migration(f);
qemu_set_fd_handler2(qemu_stdio_fd(f), NULL, NULL, NULL, NULL);
qemu_fclose(f);
}
void fd_start_incoming_migration(const char *infd, Error **errp)

19
migration-tcp.c
View File

@ -44,11 +44,8 @@ static int tcp_close(MigrationState *s)
{
int r = 0;
DPRINTF("tcp_close\n");
if (s->fd != -1) {
if (close(s->fd) < 0) {
r = -errno;
}
s->fd = -1;
if (closesocket(s->fd) < 0) {
r = -socket_error();
}
return r;
}
@ -88,12 +85,14 @@ static void tcp_accept_incoming_migration(void *opaque)
do {
c = qemu_accept(s, (struct sockaddr *)&addr, &addrlen);
} while (c == -1 && socket_error() == EINTR);
qemu_set_fd_handler2(s, NULL, NULL, NULL, NULL);
closesocket(s);
DPRINTF("accepted migration\n");
if (c == -1) {
fprintf(stderr, "could not accept migration connection\n");
goto out2;
goto out;
}
f = qemu_fopen_socket(c);
@ -103,12 +102,10 @@ static void tcp_accept_incoming_migration(void *opaque)
}
process_incoming_migration(f);
qemu_fclose(f);
return;
out:
close(c);
out2:
qemu_set_fd_handler2(s, NULL, NULL, NULL, NULL);
close(s);
closesocket(c);
}
void tcp_start_incoming_migration(const char *host_port, Error **errp)

17
migration-unix.c
View File

@ -44,11 +44,8 @@ static int unix_close(MigrationState *s)
{
int r = 0;
DPRINTF("unix_close\n");
if (s->fd != -1) {
if (close(s->fd) < 0) {
r = -errno;
}
s->fd = -1;
if (close(s->fd) < 0) {
r = -errno;
}
return r;
}
@ -88,12 +85,14 @@ static void unix_accept_incoming_migration(void *opaque)
do {
c = qemu_accept(s, (struct sockaddr *)&addr, &addrlen);
} while (c == -1 && errno == EINTR);
qemu_set_fd_handler2(s, NULL, NULL, NULL, NULL);
close(s);
DPRINTF("accepted migration\n");
if (c == -1) {
fprintf(stderr, "could not accept migration connection\n");
goto out2;
goto out;
}
f = qemu_fopen_socket(c);
@ -103,12 +102,10 @@ static void unix_accept_incoming_migration(void *opaque)
}
process_incoming_migration(f);
qemu_fclose(f);
return;
out:
close(c);
out2:
qemu_set_fd_handler2(s, NULL, NULL, NULL, NULL);
close(s);
}
void unix_start_incoming_migration(const char *path, Error **errp)

46
migration.c
View File

@ -83,9 +83,15 @@ void qemu_start_incoming_migration(const char *uri, Error **errp)
}
}
void process_incoming_migration(QEMUFile *f)
static void process_incoming_migration_co(void *opaque)
{
if (qemu_loadvm_state(f) < 0) {
QEMUFile *f = opaque;
int ret;
ret = qemu_loadvm_state(f);
qemu_set_fd_handler(qemu_get_fd(f), NULL, NULL, NULL);
qemu_fclose(f);
if (ret < 0) {
fprintf(stderr, "load of migration failed\n");
exit(0);
}
@ -103,6 +109,23 @@ void process_incoming_migration(QEMUFile *f)
}
}
static void enter_migration_coroutine(void *opaque)
{
Coroutine *co = opaque;
qemu_coroutine_enter(co, NULL);
}
void process_incoming_migration(QEMUFile *f)
{
Coroutine *co = qemu_coroutine_create(process_incoming_migration_co);
int fd = qemu_get_fd(f);
assert(fd != -1);
socket_set_nonblock(fd);
qemu_set_fd_handler(fd, enter_migration_coroutine, NULL, co);
qemu_coroutine_enter(co, f);
}
/* amount of nanoseconds we are willing to wait for migration to be down.
* the choice of nanoseconds is because it is the maximum resolution that
* get_clock() can achieve. It is an internal measure. All user-visible
@ -243,21 +266,13 @@ static int migrate_fd_cleanup(MigrationState *s)
{
int ret = 0;
if (s->fd != -1) {
qemu_set_fd_handler2(s->fd, NULL, NULL, NULL, NULL);
}
if (s->file) {
DPRINTF("closing file\n");
ret = qemu_fclose(s->file);
s->file = NULL;
}
if (s->fd != -1) {
close(s->fd);
s->fd = -1;
}
migrate_fd_close(s);
return ret;
}
@ -393,8 +408,13 @@ int migrate_fd_wait_for_unfreeze(MigrationState *s)
int migrate_fd_close(MigrationState *s)
{
qemu_set_fd_handler2(s->fd, NULL, NULL, NULL, NULL);
return s->close(s);
int rc = 0;
if (s->fd != -1) {
qemu_set_fd_handler2(s->fd, NULL, NULL, NULL, NULL);
rc = s->close(s);
s->fd = -1;
}
return rc;
}
void add_migration_state_change_notifier(Notifier *notify)

23
qemu-file.h
View File

@ -47,6 +47,10 @@ typedef int (QEMUFileGetBufferFunc)(void *opaque, uint8_t *buf,
*/
typedef int (QEMUFileCloseFunc)(void *opaque);
/* Called to return the OS file descriptor associated to the QEMUFile.
*/
typedef int (QEMUFileGetFD)(void *opaque);
/* Called to determine if the file has exceeded its bandwidth allocation. The
* bandwidth capping is a soft limit, not a hard limit.
*/
@ -59,18 +63,23 @@ typedef int (QEMUFileRateLimit)(void *opaque);
typedef int64_t (QEMUFileSetRateLimit)(void *opaque, int64_t new_rate);
typedef int64_t (QEMUFileGetRateLimit)(void *opaque);
QEMUFile *qemu_fopen_ops(void *opaque, QEMUFilePutBufferFunc *put_buffer,
QEMUFileGetBufferFunc *get_buffer,
QEMUFileCloseFunc *close,
QEMUFileRateLimit *rate_limit,
QEMUFileSetRateLimit *set_rate_limit,
QEMUFileGetRateLimit *get_rate_limit);
typedef struct QEMUFileOps {
QEMUFilePutBufferFunc *put_buffer;
QEMUFileGetBufferFunc *get_buffer;
QEMUFileCloseFunc *close;
QEMUFileGetFD *get_fd;
QEMUFileRateLimit *rate_limit;
QEMUFileSetRateLimit *set_rate_limit;
QEMUFileGetRateLimit *get_rate_limit;
} QEMUFileOps;
QEMUFile *qemu_fopen_ops(void *opaque, const QEMUFileOps *ops);
QEMUFile *qemu_fopen(const char *filename, const char *mode);
QEMUFile *qemu_fdopen(int fd, const char *mode);
QEMUFile *qemu_fopen_socket(int fd);
QEMUFile *qemu_popen(FILE *popen_file, const char *mode);
QEMUFile *qemu_popen_cmd(const char *command, const char *mode);
int qemu_stdio_fd(QEMUFile *f);
int qemu_get_fd(QEMUFile *f);
int qemu_fclose(QEMUFile *f);
void qemu_put_buffer(QEMUFile *f, const uint8_t *buf, int size);
void qemu_put_byte(QEMUFile *f, int v);

188
savevm.c
View File

@ -163,12 +163,7 @@ void qemu_announce_self(void)
#define IO_BUF_SIZE 32768
struct QEMUFile {
QEMUFilePutBufferFunc *put_buffer;
QEMUFileGetBufferFunc *get_buffer;
QEMUFileCloseFunc *close;
QEMUFileRateLimit *rate_limit;
QEMUFileSetRateLimit *set_rate_limit;
QEMUFileGetRateLimit *get_rate_limit;
const QEMUFileOps *ops;
void *opaque;
int is_write;
@ -193,28 +188,52 @@ typedef struct QEMUFileSocket
QEMUFile *file;
} QEMUFileSocket;
static int socket_get_fd(void *opaque)
{
QEMUFileSocket *s = opaque;
return s->fd;
}
static int socket_get_buffer(void *opaque, uint8_t *buf, int64_t pos, int size)
{
QEMUFileSocket *s = opaque;
ssize_t len;
do {
for (;;) {
len = qemu_recv(s->fd, buf, size, 0);
} while (len == -1 && socket_error() == EINTR);
if (len != -1) {
break;
}
if (socket_error() == EAGAIN) {
assert(qemu_in_coroutine());
qemu_coroutine_yield();
} else if (socket_error() != EINTR) {
break;
}
}
if (len == -1)
if (len == -1) {
len = -socket_error();
}
return len;
}
static int socket_close(void *opaque)
{
QEMUFileSocket *s = opaque;
closesocket(s->fd);
g_free(s);
return 0;
}
static int stdio_get_fd(void *opaque)
{
QEMUFileStdio *s = opaque;
return fileno(s->stdio_file);
}
static int stdio_put_buffer(void *opaque, const uint8_t *buf, int64_t pos, int size)
{
QEMUFileStdio *s = opaque;
@ -227,10 +246,19 @@ static int stdio_get_buffer(void *opaque, uint8_t *buf, int64_t pos, int size)
FILE *fp = s->stdio_file;
int bytes;
do {
for (;;) {
clearerr(fp);
bytes = fread(buf, 1, size, fp);
} while ((bytes == 0) && ferror(fp) && (errno == EINTR));
if (bytes != 0 || !ferror(fp)) {
break;
}
if (errno == EAGAIN) {
assert(qemu_in_coroutine());
qemu_coroutine_yield();
} else if (errno != EINTR) {
break;
}
}
return bytes;
}
@ -257,6 +285,18 @@ static int stdio_fclose(void *opaque)
return ret;
}
static const QEMUFileOps stdio_pipe_read_ops = {
.get_fd = stdio_get_fd,
.get_buffer = stdio_get_buffer,
.close = stdio_pclose
};
static const QEMUFileOps stdio_pipe_write_ops = {
.get_fd = stdio_get_fd,
.put_buffer = stdio_put_buffer,
.close = stdio_pclose
};
QEMUFile *qemu_popen(FILE *stdio_file, const char *mode)
{
QEMUFileStdio *s;
@ -271,11 +311,9 @@ QEMUFile *qemu_popen(FILE *stdio_file, const char *mode)
s->stdio_file = stdio_file;
if(mode[0] == 'r') {
s->file = qemu_fopen_ops(s, NULL, stdio_get_buffer, stdio_pclose,
NULL, NULL, NULL);
s->file = qemu_fopen_ops(s, &stdio_pipe_read_ops);
} else {
s->file = qemu_fopen_ops(s, stdio_put_buffer, NULL, stdio_pclose,
NULL, NULL, NULL);
s->file = qemu_fopen_ops(s, &stdio_pipe_write_ops);
}
return s->file;
}
@ -292,16 +330,17 @@ QEMUFile *qemu_popen_cmd(const char *command, const char *mode)
return qemu_popen(popen_file, mode);
}
int qemu_stdio_fd(QEMUFile *f)
{
QEMUFileStdio *p;
int fd;
static const QEMUFileOps stdio_file_read_ops = {
.get_fd = stdio_get_fd,
.get_buffer = stdio_get_buffer,
.close = stdio_fclose
};
p = (QEMUFileStdio *)f->opaque;
fd = fileno(p->stdio_file);
return fd;
}
static const QEMUFileOps stdio_file_write_ops = {
.get_fd = stdio_get_fd,
.put_buffer = stdio_put_buffer,
.close = stdio_fclose
};
QEMUFile *qemu_fdopen(int fd, const char *mode)
{
@ -320,11 +359,9 @@ QEMUFile *qemu_fdopen(int fd, const char *mode)
goto fail;
if(mode[0] == 'r') {
s->file = qemu_fopen_ops(s, NULL, stdio_get_buffer, stdio_fclose,
NULL, NULL, NULL);
s->file = qemu_fopen_ops(s, &stdio_file_read_ops);
} else {
s->file = qemu_fopen_ops(s, stdio_put_buffer, NULL, stdio_fclose,
NULL, NULL, NULL);
s->file = qemu_fopen_ops(s, &stdio_file_write_ops);
}
return s->file;
@ -333,31 +370,21 @@ fail:
return NULL;
}
static const QEMUFileOps socket_read_ops = {
.get_fd = socket_get_fd,
.get_buffer = socket_get_buffer,
.close = socket_close
};
QEMUFile *qemu_fopen_socket(int fd)
{
QEMUFileSocket *s = g_malloc0(sizeof(QEMUFileSocket));
s->fd = fd;
s->file = qemu_fopen_ops(s, NULL, socket_get_buffer, socket_close,
NULL, NULL, NULL);
s->file = qemu_fopen_ops(s, &socket_read_ops);
return s->file;
}
static int file_put_buffer(void *opaque, const uint8_t *buf,
int64_t pos, int size)
{
QEMUFileStdio *s = opaque;
fseek(s->stdio_file, pos, SEEK_SET);
return fwrite(buf, 1, size, s->stdio_file);
}
static int file_get_buffer(void *opaque, uint8_t *buf, int64_t pos, int size)
{
QEMUFileStdio *s = opaque;
fseek(s->stdio_file, pos, SEEK_SET);
return fread(buf, 1, size, s->stdio_file);
}
QEMUFile *qemu_fopen(const char *filename, const char *mode)
{
QEMUFileStdio *s;
@ -376,11 +403,9 @@ QEMUFile *qemu_fopen(const char *filename, const char *mode)
goto fail;
if(mode[0] == 'w') {
s->file = qemu_fopen_ops(s, file_put_buffer, NULL, stdio_fclose,
NULL, NULL, NULL);
s->file = qemu_fopen_ops(s, &stdio_file_write_ops);
} else {
s->file = qemu_fopen_ops(s, NULL, file_get_buffer, stdio_fclose,
NULL, NULL, NULL);
s->file = qemu_fopen_ops(s, &stdio_file_read_ops);
}
return s->file;
fail:
@ -405,32 +430,31 @@ static int bdrv_fclose(void *opaque)
return bdrv_flush(opaque);
}
static const QEMUFileOps bdrv_read_ops = {
.get_buffer = block_get_buffer,
.close = bdrv_fclose
};
static const QEMUFileOps bdrv_write_ops = {
.put_buffer = block_put_buffer,
.close = bdrv_fclose
};
static QEMUFile *qemu_fopen_bdrv(BlockDriverState *bs, int is_writable)
{
if (is_writable)
return qemu_fopen_ops(bs, block_put_buffer, NULL, bdrv_fclose,
NULL, NULL, NULL);
return qemu_fopen_ops(bs, NULL, block_get_buffer, bdrv_fclose, NULL, NULL, NULL);
return qemu_fopen_ops(bs, &bdrv_write_ops);
return qemu_fopen_ops(bs, &bdrv_read_ops);
}
QEMUFile *qemu_fopen_ops(void *opaque, QEMUFilePutBufferFunc *put_buffer,
QEMUFileGetBufferFunc *get_buffer,
QEMUFileCloseFunc *close,
QEMUFileRateLimit *rate_limit,
QEMUFileSetRateLimit *set_rate_limit,
QEMUFileGetRateLimit *get_rate_limit)
QEMUFile *qemu_fopen_ops(void *opaque, const QEMUFileOps *ops)
{
QEMUFile *f;
f = g_malloc0(sizeof(QEMUFile));
f->opaque = opaque;
f->put_buffer = put_buffer;
f->get_buffer = get_buffer;
f->close = close;
f->rate_limit = rate_limit;
f->set_rate_limit = set_rate_limit;
f->get_rate_limit = get_rate_limit;
f->ops = ops;
f->is_write = 0;
return f;
@ -453,11 +477,11 @@ static int qemu_fflush(QEMUFile *f)
{
int ret = 0;
if (!f->put_buffer)
if (!f->ops->put_buffer)
return 0;
if (f->is_write && f->buf_index > 0) {
ret = f->put_buffer(f->opaque, f->buf, f->buf_offset, f->buf_index);
ret = f->ops->put_buffer(f->opaque, f->buf, f->buf_offset, f->buf_index);
if (ret >= 0) {
f->buf_offset += f->buf_index;
}
@ -471,7 +495,7 @@ static void qemu_fill_buffer(QEMUFile *f)
int len;
int pending;
if (!f->get_buffer)
if (!f->ops->get_buffer)
return;
if (f->is_write)
@ -484,7 +508,7 @@ static void qemu_fill_buffer(QEMUFile *f)
f->buf_index = 0;
f->buf_size = pending;
len = f->get_buffer(f->opaque, f->buf + pending, f->buf_offset,
len = f->ops->get_buffer(f->opaque, f->buf + pending, f->buf_offset,
IO_BUF_SIZE - pending);
if (len > 0) {
f->buf_size += len;
@ -495,6 +519,14 @@ static void qemu_fill_buffer(QEMUFile *f)
qemu_file_set_error(f, len);
}
int qemu_get_fd(QEMUFile *f)
{
if (f->ops->get_fd) {
return f->ops->get_fd(f->opaque);
}
return -1;
}
/** Closes the file
*
* Returns negative error value if any error happened on previous operations or
@ -508,8 +540,8 @@ int qemu_fclose(QEMUFile *f)
int ret;
ret = qemu_fflush(f);
if (f->close) {
int ret2 = f->close(f->opaque);
if (f->ops->close) {
int ret2 = f->ops->close(f->opaque);
if (ret >= 0) {
ret = ret2;
}
@ -526,7 +558,7 @@ int qemu_fclose(QEMUFile *f)
int qemu_file_put_notify(QEMUFile *f)
{
return f->put_buffer(f->opaque, NULL, 0, 0);
return f->ops->put_buffer(f->opaque, NULL, 0, 0);
}
void qemu_put_buffer(QEMUFile *f, const uint8_t *buf, int size)
@ -673,16 +705,16 @@ static int64_t qemu_ftell(QEMUFile *f)
int qemu_file_rate_limit(QEMUFile *f)
{
if (f->rate_limit)
return f->rate_limit(f->opaque);
if (f->ops->rate_limit)
return f->ops->rate_limit(f->opaque);
return 0;
}
int64_t qemu_file_get_rate_limit(QEMUFile *f)
{
if (f->get_rate_limit)
return f->get_rate_limit(f->opaque);
if (f->ops->get_rate_limit)
return f->ops->get_rate_limit(f->opaque);
return 0;
}
@ -691,8 +723,8 @@ int64_t qemu_file_set_rate_limit(QEMUFile *f, int64_t new_rate)
{
/* any failed or completed migration keeps its state to allow probing of
* migration data, but has no associated file anymore */
if (f && f->set_rate_limit)
return f->set_rate_limit(f->opaque, new_rate);
if (f && f->ops->set_rate_limit)
return f->ops->set_rate_limit(f->opaque, new_rate);
return 0;
}