To avoid the need for duplicating implementations of virStream
drivers, provide a generic implementation that can handle any
FD based stream. This code is copied from the existing impl
in the QEMU driver, with the locking moved into the stream
impl, and addition of a read callback
The FD stream code will refuse to operate on regular files or
block devices, since those can't report EAGAIN properly when
they would block on I/O
* include/libvirt/virterror.h, include/libvirt/virterror.h: Add
VIR_FROM_STREAM error domain
* src/qemu/qemu_driver.c: Remove code obsoleted by the new
generic streams driver.
* src/fdstream.h, src/fdstream.c, src/fdstream.c,
src/libvirt_private.syms: Generic reusable FD based streams
---
.x-sc_avoid_write | 2 +-
configure.ac | 3 +-
include/libvirt/virterror.h | 3 +-
po/POTFILES.in | 1 +
src/Makefile.am | 1 +
src/fdstream.c | 503 +++++++++++++++++++++++++++++++++++++++++++
src/fdstream.h | 44 ++++
src/libvirt_private.syms | 7 +
src/qemu/qemu_driver.c | 284 +------------------------
src/util/virterror.c | 3 +
10 files changed, 569 insertions(+), 282 deletions(-)
create mode 100644 src/fdstream.c
create mode 100644 src/fdstream.h
diff --git a/.x-sc_avoid_write b/.x-sc_avoid_write
index 1f893b8..b1cfd2a 100644
--- a/.x-sc_avoid_write
+++ b/.x-sc_avoid_write
@@ -1,5 +1,5 @@
^src/libvirt\.c$
-^src/qemu/qemu_driver\.c$
+^src/fdstream\.c$
^src/qemu/qemu_monitor\.c$
^src/util/util\.c$
^src/xen/xend_internal\.c$
diff --git a/configure.ac b/configure.ac
index 6b30d90..8f55992 100644
--- a/configure.ac
+++ b/configure.ac
@@ -109,7 +109,8 @@ LIBS=$old_libs
dnl Availability of various common headers (non-fatal if missing).
AC_CHECK_HEADERS([pwd.h paths.h regex.h sys/syslimits.h sys/un.h \
- sys/poll.h syslog.h mntent.h net/ethernet.h linux/magic.h])
+ sys/poll.h syslog.h mntent.h net/ethernet.h linux/magic.h \
+ sys/un.h])
AC_CHECK_LIB([intl],[gettext],[])
diff --git a/include/libvirt/virterror.h b/include/libvirt/virterror.h
index 94d686c..abf6945 100644
--- a/include/libvirt/virterror.h
+++ b/include/libvirt/virterror.h
@@ -73,7 +73,8 @@ typedef enum {
VIR_FROM_NWFILTER, /* Error from network filter driver */
VIR_FROM_HOOK, /* Error from Synchronous hooks */
VIR_FROM_DOMAIN_SNAPSHOT, /* Error from domain snapshot */
- VIR_FROM_AUDIT /* Error from auditing subsystem */
+ VIR_FROM_AUDIT, /* Error from auditing subsystem */
+ VIR_FROM_STREAMS, /* Error from I/O streams */
} virErrorDomain;
diff --git a/po/POTFILES.in b/po/POTFILES.in
index dafef47..f0f75b0 100644
--- a/po/POTFILES.in
+++ b/po/POTFILES.in
@@ -26,6 +26,7 @@ src/esx/esx_vi.c
src/esx/esx_vi_methods.c
src/esx/esx_vi_types.c
src/esx/esx_vmx.c
+src/fdstream.c
src/interface/netcf_driver.c
src/internal.h
src/libvirt.c
diff --git a/src/Makefile.am b/src/Makefile.am
index 4a11c37..2022b85 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -94,6 +94,7 @@ DRIVER_SOURCES = \
driver.c driver.h \
internal.h \
datatypes.c datatypes.h \
+ fdstream.c fdstream.h \
$(NODE_INFO_SOURCES) \
libvirt.c libvirt_internal.h
diff --git a/src/fdstream.c b/src/fdstream.c
new file mode 100644
index 0000000..952bd69
--- /dev/null
+++ b/src/fdstream.c
@@ -0,0 +1,503 @@
+/*
+ * fdstream.h: generic streams impl for file descriptors
+ *
+ * Copyright (C) 2009-2010 Red Hat, Inc.
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+ *
+ */
+
+#include <config.h>
+
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include <sys/socket.h>
+#if HAVE_SYS_UN_H
+# include <sys/un.h>
+#endif
+#include <netinet/in.h>
+
+#include "fdstream.h"
+#include "virterror_internal.h"
+#include "datatypes.h"
+#include "memory.h"
+#include "event.h"
+#include "util.h"
+#include "files.h"
+
+#define VIR_FROM_THIS VIR_FROM_STREAMS
+#define streamsReportError(code, ...) \
+ virReportErrorHelper(NULL, VIR_FROM_THIS, code, __FILE__, \
+ __FUNCTION__, __LINE__, __VA_ARGS__)
+
+/* Tunnelled migration stream support */
+struct virFDStreamData {
+ int fd;
+
+ int watch;
+ unsigned int cbRemoved;
+ unsigned int dispatching;
+ virStreamEventCallback cb;
+ void *opaque;
+ virFreeCallback ff;
+
+ virMutex lock;
+};
+
+static int virFDStreamRemoveCallback(virStreamPtr stream)
+{
+ struct virFDStreamData *fdst = stream->privateData;
+ int ret = -1;
+
+ if (!fdst) {
+ streamsReportError(VIR_ERR_INTERNAL_ERROR,
+ "%s", _("stream is not open"));
+ return -1;
+ }
+
+ virMutexLock(&fdst->lock);
+ if (fdst->watch == 0) {
+ streamsReportError(VIR_ERR_INTERNAL_ERROR,
+ "%s", _("stream does not have a callback registered"));
+ goto cleanup;
+ }
+
+ virEventRemoveHandle(fdst->watch);
+ if (fdst->dispatching)
+ fdst->cbRemoved = 1;
+ else if (fdst->ff)
+ (fdst->ff)(fdst->opaque);
+
+ fdst->watch = 0;
+ fdst->ff = NULL;
+ fdst->cb = NULL;
+ fdst->opaque = NULL;
+
+ ret = 0;
+
+cleanup:
+ virMutexUnlock(&fdst->lock);
+ return ret;
+}
+
+static int virFDStreamUpdateCallback(virStreamPtr stream, int events)
+{
+ struct virFDStreamData *fdst = stream->privateData;
+ int ret = -1;
+
+ if (!fdst) {
+ streamsReportError(VIR_ERR_INTERNAL_ERROR,
+ "%s", _("stream is not open"));
+ return -1;
+ }
+
+ virMutexLock(&fdst->lock);
+ if (fdst->watch == 0) {
+ streamsReportError(VIR_ERR_INTERNAL_ERROR,
+ "%s", _("stream does not have a callback registered"));
+ goto cleanup;
+ }
+
+ virEventUpdateHandle(fdst->watch, events);
+
+ ret = 0;
+
+cleanup:
+ virMutexUnlock(&fdst->lock);
+ return ret;
+}
+
+static void virFDStreamEvent(int watch ATTRIBUTE_UNUSED,
+ int fd ATTRIBUTE_UNUSED,
+ int events,
+ void *opaque)
+{
+ virStreamPtr stream = opaque;
+ struct virFDStreamData *fdst = stream->privateData;
+ virStreamEventCallback cb;
+ void *cbopaque;
+ virFreeCallback ff;
+
+ if (!fdst)
+ return;
+
+ virMutexLock(&fdst->lock);
+ if (!fdst->cb) {
+ virMutexUnlock(&fdst->lock);
+ return;
+ }
+
+ cb = fdst->cb;
+ cbopaque = fdst->opaque;
+ ff = fdst->ff;
+ fdst->dispatching = 1;
+ virMutexUnlock(&fdst->lock);
+
+ cb(stream, events, cbopaque);
+
+ virMutexLock(&fdst->lock);
+ fdst->dispatching = 0;
+ if (fdst->cbRemoved && ff)
+ (ff)(cbopaque);
+ virMutexUnlock(&fdst->lock);
+}
+
+static int
+virFDStreamAddCallback(virStreamPtr st,
+ int events,
+ virStreamEventCallback cb,
+ void *opaque,
+ virFreeCallback ff)
+{
+ struct virFDStreamData *fdst = st->privateData;
+ int ret = -1;
+
+ if (!fdst) {
+ streamsReportError(VIR_ERR_INTERNAL_ERROR,
+ "%s", _("stream is not open"));
+ return -1;
+ }
+
+ virMutexLock(&fdst->lock);
+ if (fdst->watch != 0) {
+ streamsReportError(VIR_ERR_INTERNAL_ERROR,
+ "%s", _("stream already has a callback registered"));
+ goto cleanup;
+ }
+
+ if ((fdst->watch = virEventAddHandle(fdst->fd,
+ events,
+ virFDStreamEvent,
+ st,
+ NULL)) < 0) {
+ streamsReportError(VIR_ERR_INTERNAL_ERROR,
+ "%s", _("cannot register file watch on stream"));
+ goto cleanup;
+ }
+
+ fdst->cbRemoved = 0;
+ fdst->cb = cb;
+ fdst->opaque = opaque;
+ fdst->ff = ff;
+ virStreamRef(st);
+
+ ret = 0;
+
+cleanup:
+ virMutexUnlock(&fdst->lock);
+ return ret;
+}
+
+static int virFDStreamFree(struct virFDStreamData *fdst)
+{
+ int ret;
+ ret = VIR_CLOSE(fdst->fd);
+ VIR_FREE(fdst);
+ return ret;
+}
+
+
+static int
+virFDStreamClose(virStreamPtr st)
+{
+ struct virFDStreamData *fdst = st->privateData;
+ int ret;
+
+ if (!fdst)
+ return 0;
+
+ virMutexLock(&fdst->lock);
+
+ ret = virFDStreamFree(fdst);
+
+ st->privateData = NULL;
+
+ virMutexUnlock(&fdst->lock);
+
+ return ret;
+}
+
+static int virFDStreamWrite(virStreamPtr st, const char *bytes, size_t nbytes)
+{
+ struct virFDStreamData *fdst = st->privateData;
+ int ret;
+
+ if (nbytes > INT_MAX) {
+ virReportSystemError(ERANGE, "%s",
+ _("Too many bytes to write to stream"));
+ return -1;
+ }
+
+ if (!fdst) {
+ streamsReportError(VIR_ERR_INTERNAL_ERROR,
+ "%s", _("stream is not open"));
+ return -1;
+ }
+
+ virMutexLock(&fdst->lock);
+
+retry:
+ ret = write(fdst->fd, bytes, nbytes);
+ if (ret < 0) {
+ if (errno == EAGAIN || errno == EWOULDBLOCK) {
+ ret = -2;
+ } else if (errno == EINTR) {
+ goto retry;
+ } else {
+ ret = -1;
+ virReportSystemError(errno, "%s",
+ _("cannot write to stream"));
+ }
+ }
+
+ virMutexUnlock(&fdst->lock);
+ return ret;
+}
+
+
+static int virFDStreamRead(virStreamPtr st, char *bytes, size_t nbytes)
+{
+ struct virFDStreamData *fdst = st->privateData;
+ int ret;
+
+ if (nbytes > INT_MAX) {
+ virReportSystemError(ERANGE, "%s",
+ _("Too many bytes to read from stream"));
+ return -1;
+ }
+
+ if (!fdst) {
+ streamsReportError(VIR_ERR_INTERNAL_ERROR,
+ "%s", _("stream is not open"));
+ return -1;
+ }
+
+ virMutexLock(&fdst->lock);
+
+retry:
+ ret = read(fdst->fd, bytes, nbytes);
+ if (ret < 0) {
+ if (errno == EAGAIN || errno == EWOULDBLOCK) {
+ ret = -2;
+ } else if (errno == EINTR) {
+ goto retry;
+ } else {
+ ret = -1;
+ virReportSystemError(errno, "%s",
+ _("cannot read from stream"));
+ }
+ }
+
+ virMutexUnlock(&fdst->lock);
+ return ret;
+}
+
+
+static virStreamDriver virFDStreamDrv = {
+ .streamSend = virFDStreamWrite,
+ .streamRecv = virFDStreamRead,
+ .streamFinish = virFDStreamClose,
+ .streamAbort = virFDStreamClose,
+ .streamAddCallback = virFDStreamAddCallback,
+ .streamUpdateCallback = virFDStreamUpdateCallback,
+ .streamRemoveCallback = virFDStreamRemoveCallback
+};
+
+int virFDStreamOpen(virStreamPtr st,
+ int fd)
+{
+ struct virFDStreamData *fdst;
+
+ if ((st->flags & VIR_STREAM_NONBLOCK) &&
+ virSetNonBlock(fd) < 0)
+ return -1;
+
+ if (VIR_ALLOC(fdst) < 0) {
+ virReportOOMError();
+ return -1;
+ }
+
+ fdst->fd = fd;
+ if (virMutexInit(&fdst->lock) < 0) {
+ VIR_FREE(fdst);
+ streamsReportError(VIR_ERR_INTERNAL_ERROR, "%s",
+ _("Unable to initialize mutex"));
+ return -1;
+ }
+
+ st->driver = &virFDStreamDrv;
+ st->privateData = fdst;
+
+ return 0;
+}
+
+
+#if HAVE_SYS_UN_H
+int virFDStreamConnectUNIX(virStreamPtr st,
+ const char *path,
+ bool abstract)
+{
+ struct sockaddr_un sa;
+ int i = 0;
+ int timeout = 3;
+ int ret;
+
+ int fd = socket(AF_UNIX, SOCK_STREAM, 0);
+ if (fd < 0) {
+ virReportSystemError(errno, "%s", _("Unable to open UNIX socket"));
+ goto error;
+ }
+
+ memset(&sa, 0, sizeof(sa));
+ sa.sun_family = AF_UNIX;
+ if (abstract) {
+ if (virStrcpy(sa.sun_path+1, path, sizeof(sa.sun_path)-1) == NULL)
+ goto error;
+ sa.sun_path[0] = '\0';
+ } else {
+ if (virStrcpy(sa.sun_path, path, sizeof(sa.sun_path)) == NULL)
+ goto error;
+ }
+
+ do {
+ ret = connect(fd, (struct sockaddr *)&sa, sizeof(sa));
+ if (ret == 0)
+ break;
+
+ if (errno == ENOENT || errno == ECONNREFUSED) {
+ /* ENOENT : Socket may not have shown up yet
+ * ECONNREFUSED : Leftover socket hasn't been removed yet */
+ continue;
+ }
+
+ goto error;
+ } while ((++i <= timeout*5) && (usleep(.2 * 1000000) <= 0));
+
+ if (virFDStreamOpen(st, fd) < 0)
+ goto error;
+ return 0;
+
+error:
+ VIR_FORCE_CLOSE(fd);
+ return -1;
+}
+#else
+int virFDStreamConnectUNIX(virStreamPtr st ATTRIBUTE_UNUSED,
+ const char *path ATTRIBUTE_UNUSED,
+ bool abstract ATTRIBUTE_UNUSED)
+{
+ virReportSystemError(ENOSYS, "%s",
+ _("UNIX domain sockets are not supported on this platform"));
+ return -1;
+}
+#endif
+
+int virFDStreamOpenFile(virStreamPtr st,
+ const char *path,
+ int flags)
+{
+ int fd;
+ struct stat sb;
+
+ if (flags & O_CREAT) {
+ streamsReportError(VIR_ERR_INTERNAL_ERROR, "%s",
+ _("Unexpected O_CREAT flag when opening existing file"));
+ }
+
+ if ((fd = open(path, flags)) < 0) {
+ virReportSystemError(errno,
+ _("Unable to open stream for '%s'"),
+ path);
+ return -1;
+ }
+
+ if (fstat(fd, &sb) < 0) {
+ virReportSystemError(errno,
+ _("Unable to access stream for '%s'"),
+ path);
+ goto error;
+ }
+
+ /* Thanks to the POSIX i/o model, we can't reliably get
+ * non-blocking I/O on block devs/regular files. To
+ * support those we need to fork a helper process todo
+ * the I/O so we just have a fifo. Or use AIO :-(
+ */
+ if ((st->flags & VIR_STREAM_NONBLOCK) &&
+ (!S_ISCHR(sb.st_mode) &&
+ !S_ISFIFO(sb.st_mode))) {
+ streamsReportError(VIR_ERR_INTERNAL_ERROR,
+ _("Non-blocking I/O is not supported on %s"),
+ path);
+ goto error;
+ }
+
+ if (virFDStreamOpen(st, fd) < 0)
+ goto error;
+
+ return 0;
+
+error:
+ close(fd);
+ return -1;
+}
+
+int virFDStreamCreateFile(virStreamPtr st,
+ const char *path,
+ int flags,
+ mode_t mode)
+{
+ int fd = open(path, flags, mode);
+ struct stat sb;
+
+ if (fd < 0) {
+ virReportSystemError(errno,
+ _("Unable to open stream for '%s'"),
+ path);
+ return -1;
+ }
+
+ if (fstat(fd, &sb) < 0) {
+ virReportSystemError(errno,
+ _("Unable to access stream for '%s'"),
+ path);
+ goto error;
+ }
+
+ /* Thanks to the POSIX i/o model, we can't reliably get
+ * non-blocking I/O on block devs/regular files. To
+ * support those we need to fork a helper process todo
+ * the I/O so we just have a fifo. Or use AIO :-(
+ */
+ if ((st->flags & VIR_STREAM_NONBLOCK) &&
+ (!S_ISCHR(sb.st_mode) &&
+ !S_ISFIFO(sb.st_mode))) {
+ streamsReportError(VIR_ERR_INTERNAL_ERROR,
+ _("Non-blocking I/O is not supported on %s"),
+ path);
+ goto error;
+ }
+
+ if (virFDStreamOpen(st, fd) < 0)
+ goto error;
+
+ return 0;
+
+error:
+ close(fd);
+ return -1;
+}
diff --git a/src/fdstream.h b/src/fdstream.h
new file mode 100644
index 0000000..f8d22d5
--- /dev/null
+++ b/src/fdstream.h
@@ -0,0 +1,44 @@
+/*
+ * fdstream.h: generic streams impl for file descriptors
+ *
+ * Copyright (C) 2009-2010 Red Hat, Inc.
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+ *
+ */
+
+
+#ifndef __VIR_FDSTREAM_H_
+# define __VIR_FDSTREAM_H_
+
+# include "internal.h"
+# include <stdbool.h>
+
+int virFDStreamOpen(virStreamPtr st,
+ int fd);
+
+int virFDStreamConnectUNIX(virStreamPtr st,
+ const char *path,
+ bool abstract);
+
+int virFDStreamOpenFile(virStreamPtr st,
+ const char *path,
+ int flags);
+int virFDStreamCreateFile(virStreamPtr st,
+ const char *path,
+ int flags,
+ mode_t mode);
+
+#endif /* __VIR_FDSTREAM_H_ */
diff --git a/src/libvirt_private.syms b/src/libvirt_private.syms
index 430c7c1..693e2ed 100644
--- a/src/libvirt_private.syms
+++ b/src/libvirt_private.syms
@@ -331,6 +331,13 @@ virEventUpdateHandle;
virEventUpdateTimeout;
+# fdstream.h
+virFDStreamOpen;
+virFDStreamConnectUNIX;
+virFDStreamOpenFile;
+virFDStreamCreateFile;
+
+
# files.h
virClose;
diff --git a/src/qemu/qemu_driver.c b/src/qemu/qemu_driver.c
index d70ae8b..c6c3133 100644
--- a/src/qemu/qemu_driver.c
+++ b/src/qemu/qemu_driver.c
@@ -81,6 +81,7 @@
#include "hooks.h"
#include "storage_file.h"
#include "virtaudit.h"
+#include "fdstream.h"
#define VIR_FROM_THIS VIR_FROM_QEMU
@@ -10800,279 +10801,6 @@ qemuDomainIsMigratable(virDomainDefPtr def)
return true;
}
-
-/* Tunnelled migration stream support */
-struct qemuStreamMigFile {
- int fd;
-
- int watch;
- unsigned int cbRemoved;
- unsigned int dispatching;
- virStreamEventCallback cb;
- void *opaque;
- virFreeCallback ff;
-};
-
-static int qemuStreamMigRemoveCallback(virStreamPtr stream)
-{
- struct qemud_driver *driver = stream->conn->privateData;
- struct qemuStreamMigFile *qemust = stream->privateData;
- int ret = -1;
-
- if (!qemust) {
- qemuReportError(VIR_ERR_INTERNAL_ERROR,
- "%s", _("stream is not open"));
- return -1;
- }
-
- qemuDriverLock(driver);
- if (qemust->watch == 0) {
- qemuReportError(VIR_ERR_INTERNAL_ERROR,
- "%s", _("stream does not have a callback registered"));
- goto cleanup;
- }
-
- virEventRemoveHandle(qemust->watch);
- if (qemust->dispatching)
- qemust->cbRemoved = 1;
- else if (qemust->ff)
- (qemust->ff)(qemust->opaque);
-
- qemust->watch = 0;
- qemust->ff = NULL;
- qemust->cb = NULL;
- qemust->opaque = NULL;
-
- ret = 0;
-
-cleanup:
- qemuDriverUnlock(driver);
- return ret;
-}
-
-static int qemuStreamMigUpdateCallback(virStreamPtr stream, int events)
-{
- struct qemud_driver *driver = stream->conn->privateData;
- struct qemuStreamMigFile *qemust = stream->privateData;
- int ret = -1;
-
- if (!qemust) {
- qemuReportError(VIR_ERR_INTERNAL_ERROR,
- "%s", _("stream is not open"));
- return -1;
- }
-
- qemuDriverLock(driver);
- if (qemust->watch == 0) {
- qemuReportError(VIR_ERR_INTERNAL_ERROR,
- "%s", _("stream does not have a callback registered"));
- goto cleanup;
- }
-
- virEventUpdateHandle(qemust->watch, events);
-
- ret = 0;
-
-cleanup:
- qemuDriverUnlock(driver);
- return ret;
-}
-
-static void qemuStreamMigEvent(int watch ATTRIBUTE_UNUSED,
- int fd ATTRIBUTE_UNUSED,
- int events,
- void *opaque)
-{
- virStreamPtr stream = opaque;
- struct qemud_driver *driver = stream->conn->privateData;
- struct qemuStreamMigFile *qemust = stream->privateData;
- virStreamEventCallback cb;
- void *cbopaque;
- virFreeCallback ff;
-
- qemuDriverLock(driver);
- if (!qemust || !qemust->cb) {
- qemuDriverUnlock(driver);
- return;
- }
-
- cb = qemust->cb;
- cbopaque = qemust->opaque;
- ff = qemust->ff;
- qemust->dispatching = 1;
- qemuDriverUnlock(driver);
-
- cb(stream, events, cbopaque);
-
- qemuDriverLock(driver);
- qemust->dispatching = 0;
- if (qemust->cbRemoved && ff)
- (ff)(cbopaque);
- qemuDriverUnlock(driver);
-}
-
-static int
-qemuStreamMigAddCallback(virStreamPtr st,
- int events,
- virStreamEventCallback cb,
- void *opaque,
- virFreeCallback ff)
-{
- struct qemud_driver *driver = st->conn->privateData;
- struct qemuStreamMigFile *qemust = st->privateData;
- int ret = -1;
-
- if (!qemust) {
- qemuReportError(VIR_ERR_INTERNAL_ERROR,
- "%s", _("stream is not open"));
- return -1;
- }
-
- qemuDriverLock(driver);
- if (qemust->watch != 0) {
- qemuReportError(VIR_ERR_INTERNAL_ERROR,
- "%s", _("stream already has a callback registered"));
- goto cleanup;
- }
-
- if ((qemust->watch = virEventAddHandle(qemust->fd,
- events,
- qemuStreamMigEvent,
- st,
- NULL)) < 0) {
- qemuReportError(VIR_ERR_INTERNAL_ERROR,
- "%s", _("cannot register file watch on stream"));
- goto cleanup;
- }
-
- qemust->cbRemoved = 0;
- qemust->cb = cb;
- qemust->opaque = opaque;
- qemust->ff = ff;
- virStreamRef(st);
-
- ret = 0;
-
-cleanup:
- qemuDriverUnlock(driver);
- return ret;
-}
-
-static void qemuStreamMigFree(struct qemuStreamMigFile *qemust)
-{
- if (qemust->fd != -1)
- close(qemust->fd);
- VIR_FREE(qemust);
-}
-
-static struct qemuStreamMigFile *qemuStreamMigOpen(virStreamPtr st,
- const char *unixfile)
-{
- struct qemuStreamMigFile *qemust = NULL;
- struct sockaddr_un sa_qemu;
- int i = 0;
- int timeout = 3;
- int ret;
-
- if (VIR_ALLOC(qemust) < 0) {
- virReportOOMError();
- return NULL;
- }
-
- qemust->fd = socket(AF_UNIX, SOCK_STREAM, 0);
- if (qemust->fd < 0)
- goto cleanup;
-
- memset(&sa_qemu, 0, sizeof(sa_qemu));
- sa_qemu.sun_family = AF_UNIX;
- if (virStrcpy(sa_qemu.sun_path, unixfile, sizeof(sa_qemu.sun_path)) == NULL)
- goto cleanup;
-
- do {
- ret = connect(qemust->fd, (struct sockaddr *)&sa_qemu, sizeof(sa_qemu));
- if (ret == 0)
- break;
-
- if (errno == ENOENT || errno == ECONNREFUSED) {
- /* ENOENT : Socket may not have shown up yet
- * ECONNREFUSED : Leftover socket hasn't been removed yet */
- continue;
- }
-
- goto cleanup;
- } while ((++i <= timeout*5) && (usleep(.2 * 1000000) <= 0));
-
- if ((st->flags & VIR_STREAM_NONBLOCK) && virSetNonBlock(qemust->fd) < 0)
- goto cleanup;
-
- return qemust;
-
-cleanup:
- qemuStreamMigFree(qemust);
- return NULL;
-}
-
-static int
-qemuStreamMigClose(virStreamPtr st)
-{
- struct qemud_driver *driver = st->conn->privateData;
- struct qemuStreamMigFile *qemust = st->privateData;
-
- if (!qemust)
- return 0;
-
- qemuDriverLock(driver);
-
- qemuStreamMigFree(qemust);
-
- st->privateData = NULL;
-
- qemuDriverUnlock(driver);
-
- return 0;
-}
-
-static int qemuStreamMigWrite(virStreamPtr st, const char *bytes, size_t nbytes)
-{
- struct qemud_driver *driver = st->conn->privateData;
- struct qemuStreamMigFile *qemust = st->privateData;
- int ret;
-
- if (!qemust) {
- qemuReportError(VIR_ERR_INTERNAL_ERROR,
- "%s", _("stream is not open"));
- return -1;
- }
-
- qemuDriverLock(driver);
-
-retry:
- ret = write(qemust->fd, bytes, nbytes);
- if (ret < 0) {
- if (errno == EAGAIN || errno == EWOULDBLOCK) {
- ret = -2;
- } else if (errno == EINTR) {
- goto retry;
- } else {
- ret = -1;
- virReportSystemError(errno, "%s",
- _("cannot write to stream"));
- }
- }
-
- qemuDriverUnlock(driver);
- return ret;
-}
-
-static virStreamDriver qemuStreamMigDrv = {
- .streamSend = qemuStreamMigWrite,
- .streamFinish = qemuStreamMigClose,
- .streamAbort = qemuStreamMigClose,
- .streamAddCallback = qemuStreamMigAddCallback,
- .streamUpdateCallback = qemuStreamMigUpdateCallback,
- .streamRemoveCallback = qemuStreamMigRemoveCallback
-};
-
/* Prepare is the first step, and it runs on the destination host.
*
* This version starts an empty VM listening on a localhost TCP port, and
@@ -11095,7 +10823,6 @@ qemudDomainMigratePrepareTunnel(virConnectPtr dconn,
int internalret;
char *unixfile = NULL;
unsigned long long qemuCmdFlags;
- struct qemuStreamMigFile *qemust = NULL;
qemuDomainObjPrivatePtr priv = NULL;
struct timeval now;
@@ -11205,8 +10932,9 @@ qemudDomainMigratePrepareTunnel(virConnectPtr dconn,
goto endjob;
}
- qemust = qemuStreamMigOpen(st, unixfile);
- if (qemust == NULL) {
+ if (virFDStreamConnectUNIX(st,
+ unixfile,
+ false) < 0) {
qemuDomainStartAudit(vm, "migrated", false);
qemudShutdownVMDaemon(driver, vm, 0);
if (!vm->persistent) {
@@ -11220,10 +10948,8 @@ qemudDomainMigratePrepareTunnel(virConnectPtr dconn,
goto endjob;
}
- st->driver = &qemuStreamMigDrv;
- st->privateData = qemust;
-
qemuDomainStartAudit(vm, "migrated", true);
+
event = virDomainEventNewFromObj(vm,
VIR_DOMAIN_EVENT_STARTED,
VIR_DOMAIN_EVENT_STARTED_MIGRATED);
diff --git a/src/util/virterror.c b/src/util/virterror.c
index dc08aaa..2db014c 100644
--- a/src/util/virterror.c
+++ b/src/util/virterror.c
@@ -190,6 +190,9 @@ static const char *virErrorDomainName(virErrorDomain domain) {
case VIR_FROM_AUDIT:
dom = "Audit ";
break;
+ case VIR_FROM_STREAMS:
+ dom = "Streams ";
+ break;
}
return(dom);
}
--
1.7.2.3
--
libvir-list mailing list
libvir-list@xxxxxxxxxx
https://www.redhat.com/mailman/listinfo/libvir-list