[libvirt RFCv6 07/27] multifd-helper: new helper for parallel save/restore
Claudio Fontana
cfontana at suse.de
Thu May 5 12:52:21 UTC 2022
For the save direction, this helper listens on a unix socket
which QEMU connects to for multifd migration to files.
For the restore direction, this helper connects to a unix socket
QEMU listens at for multifd migration from files.
The file descriptors are passed as command line parameters.
Signed-off-by: Claudio Fontana <cfontana at suse.de>
---
po/POTFILES.in | 1 +
src/libvirt_private.syms | 1 +
src/util/meson.build | 15 +++
src/util/multifd-helper.c | 249 ++++++++++++++++++++++++++++++++++++++
src/util/virthread.c | 5 +
src/util/virthread.h | 1 +
6 files changed, 272 insertions(+)
create mode 100644 src/util/multifd-helper.c
diff --git a/po/POTFILES.in b/po/POTFILES.in
index 0d9adb0758..4efb330262 100644
--- a/po/POTFILES.in
+++ b/po/POTFILES.in
@@ -241,6 +241,7 @@
@SRCDIR at src/storage_file/storage_source_backingstore.c
@SRCDIR at src/test/test_driver.c
@SRCDIR at src/util/iohelper.c
+ at SRCDIR@src/util/multifd-helper.c
@SRCDIR at src/util/viralloc.c
@SRCDIR at src/util/virarptable.c
@SRCDIR at src/util/viraudit.c
diff --git a/src/libvirt_private.syms b/src/libvirt_private.syms
index 97bfca906b..5f2bee985e 100644
--- a/src/libvirt_private.syms
+++ b/src/libvirt_private.syms
@@ -3427,6 +3427,7 @@ virThreadCreateFull;
virThreadID;
virThreadIsSelf;
virThreadJoin;
+virThreadJoinRet;
virThreadMaxName;
virThreadSelf;
virThreadSelfID;
diff --git a/src/util/meson.build b/src/util/meson.build
index 84ef13ba32..2f8b57eabf 100644
--- a/src/util/meson.build
+++ b/src/util/meson.build
@@ -178,6 +178,11 @@ io_helper_sources = [
'virfile.c',
]
+multifd_helper_sources = [
+ 'multifd-helper.c',
+ 'virfile.c',
+]
+
virt_util_lib = static_library(
'virt_util',
[
@@ -218,6 +223,16 @@ if conf.has('WITH_LIBVIRTD')
libutil_dep,
],
}
+ virt_helpers += {
+ 'name': 'libvirt_multifd_helper',
+ 'sources': [
+ files(multifd_helper_sources),
+ dtrace_gen_headers,
+ ],
+ 'deps': [
+ libutil_dep,
+ ],
+ }
endif
util_inc_dir = include_directories('.')
diff --git a/src/util/multifd-helper.c b/src/util/multifd-helper.c
new file mode 100644
index 0000000000..59657578e2
--- /dev/null
+++ b/src/util/multifd-helper.c
@@ -0,0 +1,249 @@
+/*
+ * multifd-helper.c: listens on Unix socket to perform I/O to multiple files
+ *
+ * Copyright (C) 2022 SUSE LLC
+ *
+ * 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, see
+ * <http://www.gnu.org/licenses/>.
+ *
+ * This has been written to support QEMU multifd migration to file,
+ * allowing better use of cpu resources to speed up the save/restore.
+ */
+
+#include <config.h>
+
+#include <unistd.h>
+#include <fcntl.h>
+#include <stdlib.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+
+#include "virthread.h"
+#include "virfile.h"
+#include "virerror.h"
+#include "virstring.h"
+#include "virgettext.h"
+
+#define VIR_FROM_THIS VIR_FROM_STORAGE
+
+typedef struct _multiFdConnData multiFdConnData;
+struct _multiFdConnData {
+ int clientfd;
+ int filefd;
+ int oflags;
+ const char *path;
+ virThread tid;
+
+ off_t total;
+};
+
+typedef struct _multiFdThreadArgs multiFdThreadArgs;
+struct _multiFdThreadArgs {
+ int nchannels;
+ multiFdConnData *conn; /* contains main fd + nchannels */
+ const char *sun_path; /* unix socket name to use for the server */
+ struct sockaddr_un serv_addr;
+
+ off_t total;
+};
+
+static void clientThreadFunc(void *a)
+{
+ multiFdConnData *c = a;
+ c->total = virFileDiskCopy(c->filefd, c->path, c->clientfd, "socket");
+}
+
+static off_t waitClientThreads(multiFdConnData *conn, int n)
+{
+ int idx;
+ off_t total = 0;
+ for (idx = 0; idx < n; idx++) {
+ multiFdConnData *c = &conn[idx];
+ if (virThreadJoinRet(&c->tid) < 0) {
+ total = -1;
+ } else if (total >= 0) {
+ total += c->total;
+ }
+ if (VIR_CLOSE(c->clientfd) < 0) {
+ total = -1;
+ }
+ }
+ return total;
+}
+
+static void loadThreadFunc(void *a)
+{
+ multiFdThreadArgs *args = a;
+ int idx;
+ args->total = -1;
+
+ for (idx = 0; idx < args->nchannels + 1; idx++) {
+ /* Perform outgoing connections */
+ multiFdConnData *c = &args->conn[idx];
+ c->clientfd = socket(AF_UNIX, SOCK_STREAM, 0);
+ if (c->clientfd < 0) {
+ virReportSystemError(errno, "%s", _("loadThread: socket() failed"));
+ goto cleanup;
+ }
+ if (connect(c->clientfd, (const struct sockaddr *)&args->serv_addr,
+ sizeof(struct sockaddr_un)) < 0) {
+ virReportSystemError(errno, "%s", _("loadThread: connect() failed"));
+ goto cleanup;
+ }
+ if (virThreadCreate(&c->tid, true, &clientThreadFunc, c) < 0) {
+ virReportSystemError(errno, "%s", _("loadThread: client thread creation failed"));
+ goto cleanup;
+ }
+ }
+ args->total = waitClientThreads(args->conn, args->nchannels + 1);
+
+ cleanup:
+ for (idx = 0; idx < args->nchannels + 1; idx++) {
+ multiFdConnData *c = &args->conn[idx];
+ VIR_FORCE_CLOSE(c->clientfd);
+ }
+}
+
+static void saveThreadFunc(void *a)
+{
+ multiFdThreadArgs *args = a;
+ int idx;
+ const char buf[1] = {'R'};
+ int sockfd;
+
+ if ((sockfd = socket(AF_UNIX, SOCK_STREAM, 0)) < 0) {
+ virReportSystemError(errno, "%s", _("saveThread: socket() failed"));
+ return;
+ }
+ unlink(args->sun_path);
+ if (bind(sockfd, (struct sockaddr *)&args->serv_addr, sizeof(args->serv_addr)) < 0) {
+ virReportSystemError(errno, "%s", _("saveThread: bind() failed"));
+ goto cleanup;
+ }
+ if (listen(sockfd, args->nchannels + 1) < 0) {
+ virReportSystemError(errno, "%s", _("saveThread: listen() failed"));
+ goto cleanup;
+ }
+
+ /* signal that the server is ready */
+ if (safewrite(STDOUT_FILENO, &buf, 1) != 1) {
+ virReportSystemError(errno, "%s", _("saveThread: safewrite failed"));
+ goto cleanup;
+ }
+
+ for (idx = 0; idx < args->nchannels + 1; idx++) {
+ /* Wait for incoming connection. */
+ multiFdConnData *c = &args->conn[idx];
+ if ((c->clientfd = accept(sockfd, NULL, NULL)) < 0) {
+ virReportSystemError(errno, "%s", _("saveThread: accept() failed"));
+ goto cleanup;
+ }
+ if (virThreadCreate(&c->tid, true, &clientThreadFunc, c) < 0) {
+ virReportSystemError(errno, "%s", _("saveThread: client thread creation failed"));
+ goto cleanup;
+ }
+ }
+
+ args->total = waitClientThreads(args->conn, args->nchannels + 1);
+
+ cleanup:
+ for (idx = 0; idx < args->nchannels + 1; idx++) {
+ multiFdConnData *c = &args->conn[idx];
+ VIR_FORCE_CLOSE(c->clientfd);
+ }
+ if (VIR_CLOSE(sockfd) < 0)
+ args->total = -1;
+}
+
+static const char *program_name;
+
+G_GNUC_NORETURN static void
+usage(int status)
+{
+ if (status) {
+ fprintf(stderr, _("%s: try --help for more details"), program_name);
+ } else {
+ fprintf(stderr, _("Usage: %s UNIX_SOCNAME N MAINFD FD0 FD1 ... FDn"), program_name);
+ }
+ exit(status);
+}
+
+int
+main(int argc, char **argv)
+{
+ virThread tid;
+ virThreadFunc func = saveThreadFunc;
+ multiFdThreadArgs args = { 0 };
+ int idx;
+
+ /* sleep(10); */
+
+ program_name = argv[0];
+
+ if (virGettextInitialize() < 0 ||
+ virErrorInitialize() < 0) {
+ fprintf(stderr, _("%s: initialization failed"), program_name);
+ exit(EXIT_FAILURE);
+ }
+
+ if (argc > 1 && STREQ(argv[1], "--help"))
+ usage(EXIT_SUCCESS);
+ if (argc < 4)
+ usage(EXIT_FAILURE);
+
+ args.sun_path = argv[1];
+ if (virStrToLong_i(argv[2], NULL, 10, &args.nchannels) < 0)
+ fprintf(stderr, _("%s: malformed number of channels N %s"), program_name, argv[2]);
+
+ if (argc < 4 + args.nchannels)
+ usage(EXIT_FAILURE);
+
+ args.conn = g_new0(multiFdConnData, args.nchannels + 1);
+
+ for (idx = 3; idx < 3 + args.nchannels + 1; idx++) {
+ multiFdConnData *c = &args.conn[idx - 3];
+
+ if (virStrToLong_i(argv[idx], NULL, 10, &c->filefd) < 0) {
+ fprintf(stderr, _("%s: malformed FD %s"), program_name, argv[idx]);
+ usage(EXIT_FAILURE);
+ }
+#ifndef F_GETFL
+#error "multifd-helper requires F_GETFL parameter of fcntl"
+#endif
+ c->oflags = fcntl(c->filefd, F_GETFL);
+ if ((c->oflags & O_ACCMODE) == O_RDONLY) {
+ func = loadThreadFunc;
+ }
+ }
+
+ /* initialize server address structure */
+ memset(&args.serv_addr, 0, sizeof(args.serv_addr));
+ args.serv_addr.sun_family = AF_UNIX;
+ virStrcpyStatic(args.serv_addr.sun_path, args.sun_path);
+
+ if (virThreadCreate(&tid, true, func, &args) < 0) {
+ virReportSystemError(errno, _("%s: failed to create server thread"), program_name);
+ exit(EXIT_FAILURE);
+ }
+
+ if (virThreadJoinRet(&tid) < 0)
+ exit(EXIT_FAILURE);
+
+ if (args.total < 0)
+ exit(EXIT_FAILURE);
+
+ exit(EXIT_SUCCESS);
+}
diff --git a/src/util/virthread.c b/src/util/virthread.c
index 5422bb74fd..0f6c6a68fa 100644
--- a/src/util/virthread.c
+++ b/src/util/virthread.c
@@ -348,6 +348,11 @@ void virThreadJoin(virThread *thread)
pthread_join(thread->thread, NULL);
}
+int virThreadJoinRet(virThread *thread)
+{
+ return pthread_join(thread->thread, NULL);
+}
+
void virThreadCancel(virThread *thread)
{
pthread_cancel(thread->thread);
diff --git a/src/util/virthread.h b/src/util/virthread.h
index 23abe0b6c9..5cecb9bd8a 100644
--- a/src/util/virthread.h
+++ b/src/util/virthread.h
@@ -89,6 +89,7 @@ int virThreadCreateFull(virThread *thread,
void virThreadSelf(virThread *thread);
bool virThreadIsSelf(virThread *thread);
void virThreadJoin(virThread *thread);
+int virThreadJoinRet(virThread *thread);
size_t virThreadMaxName(void);
--
2.35.3
More information about the libvir-list
mailing list