11:39 p.m.
This patch series adds a new watchdog action `dump' which lets libvirtd
can do auto-dump when receiving a watchdog event from qemu guest.
In order to make the function work, there must be a watchdog device
added to guest, and guest must have a watchdog daemon running, for
example, /etc/init.d/watchdog start or auto-started on boot.
Changes:
v5:
- qemu_driver is passed into threadpool as opaque parameter rather than
visit the global qemu_driver in worker function
- same situation as above of server in libvirtd.c
- also list auto_dump_path in src/qemu/libvirtd_qemu.aug and
src/qemu/test_libvirtd_qemu.aug
- check return value of qemuDomainObjEndJob for safety
v4:
- updated threadpool to follow libvirt naming style, use appropriate
internals APIs, and hide the struct definitions from the header
(by Daniel)
- fix an error that qemuDomainObjBeginJobWithDriver() get lost in
qemuDomainCoreDump()
- use thread pool in libvirtd (qemud worker)
v3:
- let default auto-dump dir be /var/lib/libvirt/qemu/dump
Hu Tao (4):
threadpool impl
Add a new function doCoreDump
Add a watchdog action `dump'
Using threadpool API to manage qemud worker
cfg.mk | 1 +
daemon/libvirtd.c | 172 +++++------------------------
daemon/libvirtd.h | 2 +
src/Makefile.am | 1 +
src/conf/domain_conf.c | 1 +
src/conf/domain_conf.h | 1 +
src/libvirt_private.syms | 6 +
src/qemu/libvirtd_qemu.aug | 1 +
src/qemu/qemu.conf | 5 +
src/qemu/qemu_conf.c | 16 +++-
src/qemu/qemu_conf.h | 5 +
src/qemu/qemu_driver.c | 228 ++++++++++++++++++++++++++++----------
src/qemu/test_libvirtd_qemu.aug | 2 +
src/util/threadpool.c | 231 +++++++++++++++++++++++++++++++++++++++
src/util/threadpool.h | 48 ++++++++
15 files changed, 518 insertions(+), 202 deletions(-)
create mode 100644 src/util/threadpool.c
create mode 100644 src/util/threadpool.h
--
1.7.3
--
Thanks,
Hu Tao
11:39 p.m.
New subject: [libvirt] [PATCH v5 1/4] threadpool impl
* src/util/threadpool.c, src/util/threadpool.h: Thread pool
implementation
* src/Makefile.am: Build thread pool
* src/libvirt_private.syms: Export public functions
---
cfg.mk | 1 +
src/Makefile.am | 1 +
src/libvirt_private.syms | 6 +
src/util/threadpool.c | 231 ++++++++++++++++++++++++++++++++++++++++++++++
src/util/threadpool.h | 48 ++++++++++
5 files changed, 287 insertions(+), 0 deletions(-)
create mode 100644 src/util/threadpool.c
create mode 100644 src/util/threadpool.h
diff --git a/cfg.mk b/cfg.mk
index 6e474c4..0af26d2 100644
--- a/cfg.mk
+++ b/cfg.mk
@@ -127,6 +127,7 @@ useless_free_options = \
--name=virStoragePoolObjFree \
--name=virStoragePoolSourceFree \
--name=virStorageVolDefFree \
+ --name=virThreadPoolFree \
--name=xmlFree \
--name=xmlXPathFreeContext \
--name=xmlXPathFreeObject
diff --git a/src/Makefile.am b/src/Makefile.am
index a9a1986..d71c644 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -73,6 +73,7 @@ UTIL_SOURCES = \
util/threads.c util/threads.h \
util/threads-pthread.h \
util/threads-win32.h \
+ util/threadpool.c util/threadpool.h \
util/uuid.c util/uuid.h \
util/util.c util/util.h \
util/xml.c util/xml.h \
diff --git a/src/libvirt_private.syms b/src/libvirt_private.syms
index a21928a..e7b500c 100644
--- a/src/libvirt_private.syms
+++ b/src/libvirt_private.syms
@@ -862,3 +862,9 @@ virXPathStringLimit;
virXPathULong;
virXPathULongHex;
virXPathULongLong;
+
+
+# threadpool.h
+virThreadPoolNew;
+virThreadPoolFree;
+virThreadPoolSendJob;
diff --git a/src/util/threadpool.c b/src/util/threadpool.c
new file mode 100644
index 0000000..8217591
--- /dev/null
+++ b/src/util/threadpool.c
@@ -0,0 +1,231 @@
+/*
+ * threadpool.c: a generic thread pool implementation
+ *
+ * Copyright (C) 2010 Hu Tao
+ * Copyright (C) 2010 Daniel P. Berrange
+ *
+ * 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
+ *
+ * Authors:
+ * Hu Tao <hutao(a)cn.fujitsu.com>
+ * Daniel P. Berrange <berrange(a)redhat.com>
+ */
+
+#include <config.h>
+
+#include "threadpool.h"
+#include "memory.h"
+#include "threads.h"
+#include "virterror_internal.h"
+#include "ignore-value.h"
+
+#define VIR_FROM_THIS VIR_FROM_NONE
+
+typedef struct _virThreadPoolJob virThreadPoolJob;
+typedef virThreadPoolJob *virThreadPoolJobPtr;
+
+struct _virThreadPoolJob {
+ virThreadPoolJobPtr next;
+
+ void *data;
+};
+
+typedef struct _virThreadPoolJobList virThreadPoolJobList;
+typedef virThreadPoolJobList *virThreadPoolJobListPtr;
+
+struct _virThreadPoolJobList {
+ virThreadPoolJobPtr head;
+ virThreadPoolJobPtr *tail;
+};
+
+
+struct _virThreadPool {
+ bool quit;
+
+ virThreadPoolJobFunc jobFunc;
+ void *jobOpaque;
+ virThreadPoolJobList jobList;
+
+ virMutex mutex;
+ virCond cond;
+ virCond quit_cond;
+
+ size_t maxWorkers;
+ size_t freeWorkers;
+ size_t nWorkers;
+ virThreadPtr workers;
+};
+
+static void virThreadPoolWorker(void *opaque)
+{
+ virThreadPoolPtr pool = opaque;
+
+ virMutexLock(&pool->mutex);
+
+ while (1) {
+ while (!pool->quit &&
+ !pool->jobList.head) {
+ pool->freeWorkers++;
+ if (virCondWait(&pool->cond, &pool->mutex) < 0) {
+ pool->freeWorkers--;
+ goto out;
+ }
+ pool->freeWorkers--;
+ }
+
+ if (pool->quit)
+ break;
+
+ virThreadPoolJobPtr job = pool->jobList.head;
+ pool->jobList.head = pool->jobList.head->next;
+ job->next = NULL;
+ if (pool->jobList.tail == &job->next)
+ pool->jobList.tail = &pool->jobList.head;
+
+ virMutexUnlock(&pool->mutex);
+ (pool->jobFunc)(job->data, pool->jobOpaque);
+ VIR_FREE(job);
+ virMutexLock(&pool->mutex);
+ }
+
+out:
+ pool->nWorkers--;
+ if (pool->nWorkers == 0)
+ virCondSignal(&pool->quit_cond);
+ virMutexUnlock(&pool->mutex);
+}
+
+virThreadPoolPtr virThreadPoolNew(size_t minWorkers,
+ size_t maxWorkers,
+ virThreadPoolJobFunc func,
+ void *opaque)
+{
+ virThreadPoolPtr pool;
+ size_t i;
+
+ if (minWorkers > maxWorkers)
+ minWorkers = maxWorkers;
+
+ if (VIR_ALLOC(pool) < 0) {
+ virReportOOMError();
+ return NULL;
+ }
+
+ pool->jobList.head = NULL;
+ pool->jobList.tail = &pool->jobList.head;
+
+ pool->jobFunc = func;
+ pool->jobOpaque = opaque;
+
+ if (virMutexInit(&pool->mutex) < 0)
+ goto error;
+ if (virCondInit(&pool->cond) < 0)
+ goto error;
+ if (virCondInit(&pool->quit_cond) < 0)
+ goto error;
+
+ if (VIR_ALLOC_N(pool->workers, minWorkers) < 0)
+ goto error;
+
+ pool->maxWorkers = maxWorkers;
+ for (i = 0; i < minWorkers; i++) {
+ if (virThreadCreate(&pool->workers[i],
+ true,
+ virThreadPoolWorker,
+ pool) < 0) {
+ goto error;
+ }
+ pool->nWorkers++;
+ }
+
+ return pool;
+
+error:
+ virThreadPoolFree(pool);
+ return NULL;
+
+}
+
+void virThreadPoolFree(virThreadPoolPtr pool)
+{
+ virThreadPoolJobPtr job;
+
+ if (!pool)
+ return;
+
+ virMutexLock(&pool->mutex);
+ pool->quit = true;
+ if (pool->nWorkers > 0) {
+ virCondBroadcast(&pool->cond);
+ ignore_value(virCondWait(&pool->quit_cond, &pool->mutex));
+ }
+
+ while ((job = pool->jobList.head)) {
+ pool->jobList.head = pool->jobList.head->next;
+ VIR_FREE(job);
+ }
+
+ VIR_FREE(pool->workers);
+ virMutexUnlock(&pool->mutex);
+ virMutexDestroy(&pool->mutex);
+ ignore_value(virCondDestroy(&pool->quit_cond));
+ ignore_value(virCondDestroy(&pool->cond));
+ VIR_FREE(pool);
+}
+
+int virThreadPoolSendJob(virThreadPoolPtr pool,
+ void *jobData)
+{
+ virThreadPoolJobPtr job;
+
+ virMutexLock(&pool->mutex);
+ if (pool->quit)
+ goto error;
+
+ if (pool->freeWorkers == 0 &&
+ pool->nWorkers < pool->maxWorkers) {
+ if (VIR_EXPAND_N(pool->workers, pool->nWorkers, 1) < 0) {
+ virReportOOMError();
+ goto error;
+ }
+
+ if (virThreadCreate(&pool->workers[pool->nWorkers - 1],
+ true,
+ virThreadPoolWorker,
+ pool) < 0) {
+ pool->nWorkers--;
+ goto error;
+ }
+ }
+
+ if (VIR_ALLOC(job) < 0) {
+ virReportOOMError();
+ goto error;
+ }
+
+ job->data = jobData;
+ job->next = NULL;
+ *pool->jobList.tail = job;
+ pool->jobList.tail = &(*pool->jobList.tail)->next;
+
+ virCondSignal(&pool->cond);
+ virMutexUnlock(&pool->mutex);
+
+ return 0;
+
+error:
+ virMutexUnlock(&pool->mutex);
+ return -1;
+}
diff --git a/src/util/threadpool.h b/src/util/threadpool.h
new file mode 100644
index 0000000..adab68f
--- /dev/null
+++ b/src/util/threadpool.h
@@ -0,0 +1,48 @@
+/*
+ * threadpool.h: a generic thread pool implementation
+ *
+ * Copyright (C) 2010 Hu Tao
+ * Copyright (C) 2010 Daniel P. Berrange
+ *
+ * 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
+ *
+ * Author:
+ * Hu Tao <hutao(a)cn.fujitsu.com>
+ * Daniel P. Berrange <berrange(a)redhat.com>
+ */
+
+#ifndef __VIR_THREADPOOL_H__
+#define __VIR_THREADPOOL_H__
+
+#include "internal.h"
+
+typedef struct _virThreadPool virThreadPool;
+typedef virThreadPool *virThreadPoolPtr;
+
+typedef void (*virThreadPoolJobFunc)(void *jobdata, void *opaque);
+
+virThreadPoolPtr virThreadPoolNew(size_t minWorkers,
+ size_t maxWorkers,
+ virThreadPoolJobFunc func,
+ void *opaque) ATTRIBUTE_NONNULL(3);
+
+void virThreadPoolFree(virThreadPoolPtr pool);
+
+int virThreadPoolSendJob(virThreadPoolPtr pool,
+ void *jobdata) ATTRIBUTE_NONNULL(1)
+ ATTRIBUTE_NONNULL(2)
+ ATTRIBUTE_RETURN_CHECK;
+
+#endif
--
1.7.3
--
Thanks,
Hu Tao
11:39 p.m.
New subject: [libvirt] [PATCH v5 2/4] Add a new function doCoreDump
This patch prepares for the next patch.
---
src/qemu/qemu_driver.c | 132 +++++++++++++++++++++++++++---------------------
1 files changed, 74 insertions(+), 58 deletions(-)
diff --git a/src/qemu/qemu_driver.c b/src/qemu/qemu_driver.c
index 1a7c1ad..e534195 100644
--- a/src/qemu/qemu_driver.c
+++ b/src/qemu/qemu_driver.c
@@ -6063,6 +6063,78 @@ cleanup:
return ret;
}
+static int doCoreDump(struct qemud_driver *driver,
+ virDomainObjPtr vm,
+ const char *path,
+ enum qemud_save_formats compress)
+{
+ int fd = -1;
+ int ret = -1;
+ qemuDomainObjPrivatePtr priv;
+
+ priv = vm->privateData;
+
+ /* Create an empty file with appropriate ownership. */
+ if ((fd = open(path, O_CREAT|O_TRUNC|O_WRONLY, S_IRUSR|S_IWUSR)) < 0) {
+ qemuReportError(VIR_ERR_OPERATION_FAILED,
+ _("failed to create '%s'"), path);
+ goto cleanup;
+ }
+
+ if (VIR_CLOSE(fd) < 0) {
+ virReportSystemError(errno,
+ _("unable to save file %s"),
+ path);
+ goto cleanup;
+ }
+
+ if (driver->securityDriver &&
+ driver->securityDriver->domainSetSavedStateLabel &&
+
driver->securityDriver->domainSetSavedStateLabel(driver->securityDriver,
+ vm, path) == -1)
+ goto cleanup;
+
+ qemuDomainObjEnterMonitorWithDriver(driver, vm);
+ if (compress == QEMUD_SAVE_FORMAT_RAW) {
+ const char *args[] = {
+ "cat",
+ NULL,
+ };
+ ret = qemuMonitorMigrateToFile(priv->mon,
+ QEMU_MONITOR_MIGRATE_BACKGROUND,
+ args, path, 0);
+ } else {
+ const char *prog = qemudSaveCompressionTypeToString(compress);
+ const char *args[] = {
+ prog,
+ "-c",
+ NULL,
+ };
+ ret = qemuMonitorMigrateToFile(priv->mon,
+ QEMU_MONITOR_MIGRATE_BACKGROUND,
+ args, path, 0);
+ }
+ qemuDomainObjExitMonitorWithDriver(driver, vm);
+ if (ret < 0)
+ goto cleanup;
+
+ ret = qemuDomainWaitForMigrationComplete(driver, vm);
+
+ if (ret < 0)
+ goto cleanup;
+
+ if (driver->securityDriver &&
+ driver->securityDriver->domainRestoreSavedStateLabel &&
+
driver->securityDriver->domainRestoreSavedStateLabel(driver->securityDriver,
+ vm, path) == -1)
+ goto cleanup;
+
+cleanup:
+ if (ret != 0)
+ unlink(path);
+ return ret;
+}
+
static enum qemud_save_formats
getCompressionType(struct qemud_driver *driver)
{
@@ -6097,13 +6169,10 @@ static int qemudDomainCoreDump(virDomainPtr dom,
struct qemud_driver *driver = dom->conn->privateData;
virDomainObjPtr vm;
int resume = 0, paused = 0;
- int ret = -1, fd = -1;
+ int ret = -1;
virDomainEventPtr event = NULL;
- enum qemud_save_formats compress;
qemuDomainObjPrivatePtr priv;
- compress = getCompressionType(driver);
-
qemuDriverLock(driver);
vm = virDomainFindByUUID(&driver->domains, dom->uuid);
@@ -6125,26 +6194,6 @@ static int qemudDomainCoreDump(virDomainPtr dom,
goto endjob;
}
- /* Create an empty file with appropriate ownership. */
- if ((fd = open(path, O_CREAT|O_TRUNC|O_WRONLY, S_IRUSR|S_IWUSR)) < 0) {
- qemuReportError(VIR_ERR_OPERATION_FAILED,
- _("failed to create '%s'"), path);
- goto endjob;
- }
-
- if (VIR_CLOSE(fd) < 0) {
- virReportSystemError(errno,
- _("unable to save file %s"),
- path);
- goto endjob;
- }
-
- if (driver->securityDriver &&
- driver->securityDriver->domainSetSavedStateLabel &&
-
driver->securityDriver->domainSetSavedStateLabel(driver->securityDriver,
- vm, path) == -1)
- goto endjob;
-
/* Migrate will always stop the VM, so the resume condition is
independent of whether the stop command is issued. */
resume = (vm->state == VIR_DOMAIN_RUNNING);
@@ -6168,43 +6217,12 @@ static int qemudDomainCoreDump(virDomainPtr dom,
}
}
- qemuDomainObjEnterMonitorWithDriver(driver, vm);
- if (compress == QEMUD_SAVE_FORMAT_RAW) {
- const char *args[] = {
- "cat",
- NULL,
- };
- ret = qemuMonitorMigrateToFile(priv->mon,
- QEMU_MONITOR_MIGRATE_BACKGROUND,
- args, path, 0);
- } else {
- const char *prog = qemudSaveCompressionTypeToString(compress);
- const char *args[] = {
- prog,
- "-c",
- NULL,
- };
- ret = qemuMonitorMigrateToFile(priv->mon,
- QEMU_MONITOR_MIGRATE_BACKGROUND,
- args, path, 0);
- }
- qemuDomainObjExitMonitorWithDriver(driver, vm);
- if (ret < 0)
- goto endjob;
-
- ret = qemuDomainWaitForMigrationComplete(driver, vm);
-
+ ret = doCoreDump(driver, vm, path, getCompressionType(driver));
if (ret < 0)
goto endjob;
paused = 1;
- if (driver->securityDriver &&
- driver->securityDriver->domainRestoreSavedStateLabel &&
-
driver->securityDriver->domainRestoreSavedStateLabel(driver->securityDriver,
- vm, path) == -1)
- goto endjob;
-
endjob:
if ((ret == 0) && (flags & VIR_DUMP_CRASH)) {
qemudShutdownVMDaemon(driver, vm, 0);
@@ -6237,8 +6255,6 @@ endjob:
}
cleanup:
- if (ret != 0)
- unlink(path);
if (vm)
virDomainObjUnlock(vm);
if (event)
--
1.7.3
--
Thanks,
Hu Tao
11:40 p.m.
New subject: [libvirt] [PATCH v5 3/4] Add a watchdog action `dump'
`dump' watchdog action lets libvirtd to dump the guest when receives a
watchdog event (which probably means a guest crash)
Currently only qemu is supported.
---
src/conf/domain_conf.c | 1 +
src/conf/domain_conf.h | 1 +
src/qemu/libvirtd_qemu.aug | 1 +
src/qemu/qemu.conf | 5 ++
src/qemu/qemu_conf.c | 16 ++++++-
src/qemu/qemu_conf.h | 5 ++
src/qemu/qemu_driver.c | 96 +++++++++++++++++++++++++++++++++++++++
src/qemu/test_libvirtd_qemu.aug | 2 +
8 files changed, 126 insertions(+), 1 deletions(-)
diff --git a/src/conf/domain_conf.c b/src/conf/domain_conf.c
index 3f14cee..a6cb444 100644
--- a/src/conf/domain_conf.c
+++ b/src/conf/domain_conf.c
@@ -245,6 +245,7 @@ VIR_ENUM_IMPL(virDomainWatchdogAction,
VIR_DOMAIN_WATCHDOG_ACTION_LAST,
"shutdown",
"poweroff",
"pause",
+ "dump",
"none")
VIR_ENUM_IMPL(virDomainVideo, VIR_DOMAIN_VIDEO_TYPE_LAST,
diff --git a/src/conf/domain_conf.h b/src/conf/domain_conf.h
index 899b19f..7f50b79 100644
--- a/src/conf/domain_conf.h
+++ b/src/conf/domain_conf.h
@@ -462,6 +462,7 @@ enum virDomainWatchdogAction {
VIR_DOMAIN_WATCHDOG_ACTION_SHUTDOWN,
VIR_DOMAIN_WATCHDOG_ACTION_POWEROFF,
VIR_DOMAIN_WATCHDOG_ACTION_PAUSE,
+ VIR_DOMAIN_WATCHDOG_ACTION_DUMP,
VIR_DOMAIN_WATCHDOG_ACTION_NONE,
VIR_DOMAIN_WATCHDOG_ACTION_LAST
diff --git a/src/qemu/libvirtd_qemu.aug b/src/qemu/libvirtd_qemu.aug
index 78852f3..2f37015 100644
--- a/src/qemu/libvirtd_qemu.aug
+++ b/src/qemu/libvirtd_qemu.aug
@@ -37,6 +37,7 @@ module Libvirtd_qemu =
| str_array_entry "cgroup_device_acl"
| str_entry "save_image_format"
| str_entry "dump_image_format"
+ | str_entry "auto_dump_path"
| str_entry "hugetlbfs_mount"
| bool_entry "relaxed_acs_check"
| bool_entry "vnc_allow_host_audio"
diff --git a/src/qemu/qemu.conf b/src/qemu/qemu.conf
index f4f965e..ba41f80 100644
--- a/src/qemu/qemu.conf
+++ b/src/qemu/qemu.conf
@@ -191,6 +191,11 @@
# save_image_format = "raw"
# dump_image_format = "raw"
+# When a domain is configured to be auto-dumped when libvirtd receives a
+# watchdog event from qemu guest, libvirtd will save dump files in directory
+# specified by auto_dump_path. Default value is /var/lib/libvirt/qemu/dump
+#
+# auto_dump_path = "/var/lib/libvirt/qemu/dump"
# If provided by the host and a hugetlbfs mount point is configured,
# a guest may request huge page backing. When this mount point is
diff --git a/src/qemu/qemu_conf.c b/src/qemu/qemu_conf.c
index 7cd0603..187e206 100644
--- a/src/qemu/qemu_conf.c
+++ b/src/qemu/qemu_conf.c
@@ -386,6 +386,17 @@ int qemudLoadDriverConfig(struct qemud_driver *driver,
}
}
+ p = virConfGetValue (conf, "auto_dump_path");
+ CHECK_TYPE ("auto_dump_path", VIR_CONF_STRING);
+ if (p && p->str) {
+ VIR_FREE(driver->autoDumpPath);
+ if (!(driver->autoDumpPath = strdup(p->str))) {
+ virReportOOMError();
+ virConfFree(conf);
+ return -1;
+ }
+ }
+
p = virConfGetValue (conf, "hugetlbfs_mount");
CHECK_TYPE ("hugetlbfs_mount", VIR_CONF_STRING);
if (p && p->str) {
@@ -5374,7 +5385,10 @@ int qemudBuildCommandLine(virConnectPtr conn,
}
ADD_ARG(optstr);
- const char *action = virDomainWatchdogActionTypeToString(watchdog->action);
+ int act = watchdog->action;
+ if (act == VIR_DOMAIN_WATCHDOG_ACTION_DUMP)
+ act = VIR_DOMAIN_WATCHDOG_ACTION_PAUSE;
+ const char *action = virDomainWatchdogActionTypeToString(act);
if (!action) {
qemuReportError(VIR_ERR_INTERNAL_ERROR,
"%s", _("invalid watchdog action"));
diff --git a/src/qemu/qemu_conf.h b/src/qemu/qemu_conf.h
index aba64d6..9bcae88 100644
--- a/src/qemu/qemu_conf.h
+++ b/src/qemu/qemu_conf.h
@@ -41,6 +41,7 @@
# include "driver.h"
# include "bitmap.h"
# include "macvtap.h"
+# include "threadpool.h"
# define qemudDebug(fmt, ...) do {} while(0)
@@ -107,6 +108,8 @@ enum qemud_cmd_flags {
struct qemud_driver {
virMutex lock;
+ virThreadPoolPtr workerPool;
+
int privileged;
uid_t user;
@@ -174,6 +177,8 @@ struct qemud_driver {
char *saveImageFormat;
char *dumpImageFormat;
+ char *autoDumpPath;
+
pciDeviceList *activePciHostdevs;
virBitmapPtr reservedVNCPorts;
diff --git a/src/qemu/qemu_driver.c b/src/qemu/qemu_driver.c
index e534195..713179b 100644
--- a/src/qemu/qemu_driver.c
+++ b/src/qemu/qemu_driver.c
@@ -85,6 +85,7 @@
#include "files.h"
#include "fdstream.h"
#include "configmake.h"
+#include "threadpool.h"
#define VIR_FROM_THIS VIR_FROM_QEMU
@@ -137,6 +138,14 @@ struct _qemuDomainObjPrivate {
int persistentAddrs;
};
+struct watchdogEvent
+{
+ virDomainObjPtr vm;
+ int action;
+};
+
+static void processWatchdogEvent(void *data, void *opaque);
+
static int qemudShutdown(void);
static void qemuDriverLock(struct qemud_driver *driver)
@@ -1204,6 +1213,17 @@ qemuHandleDomainWatchdog(qemuMonitorPtr mon ATTRIBUTE_UNUSED,
if (virDomainSaveStatus(driver->caps, driver->stateDir, vm) < 0)
VIR_WARN("Unable to save status on vm %s after IO error",
vm->def->name);
}
+
+ if (vm->def->watchdog->action == VIR_DOMAIN_WATCHDOG_ACTION_DUMP) {
+ struct watchdogEvent *wdEvent;
+ if (VIR_ALLOC(wdEvent) == 0) {
+ wdEvent->action = VIR_DOMAIN_WATCHDOG_ACTION_DUMP;
+ wdEvent->vm = vm;
+ ignore_value(virThreadPoolSendJob(driver->workerPool, wdEvent));
+ } else
+ virReportOOMError();
+ }
+
virDomainObjUnlock(vm);
if (watchdogEvent || lifecycleEvent) {
@@ -1786,6 +1806,9 @@ qemudStartup(int privileged) {
if (virAsprintf(&qemu_driver->snapshotDir,
"%s/lib/libvirt/qemu/snapshot", LOCALSTATEDIR) == -1)
goto out_of_memory;
+ if (virAsprintf(&qemu_driver->autoDumpPath,
+ "%s/lib/libvirt/qemu/dump", LOCALSTATEDIR) == -1)
+ goto out_of_memory;
} else {
uid_t uid = geteuid();
char *userdir = virGetUserDirectory(uid);
@@ -1814,6 +1837,8 @@ qemudStartup(int privileged) {
goto out_of_memory;
if (virAsprintf(&qemu_driver->snapshotDir, "%s/qemu/snapshot",
base) == -1)
goto out_of_memory;
+ if (virAsprintf(&qemu_driver->autoDumpPath, "%s/qemu/dump",
base) == -1)
+ goto out_of_memory;
}
if (virFileMakePath(qemu_driver->stateDir) != 0) {
@@ -1846,6 +1871,12 @@ qemudStartup(int privileged) {
qemu_driver->snapshotDir, virStrerror(errno, ebuf, sizeof ebuf));
goto error;
}
+ if (virFileMakePath(qemu_driver->autoDumpPath) != 0) {
+ char ebuf[1024];
+ VIR_ERROR(_("Failed to create dump dir '%s': %s"),
+ qemu_driver->autoDumpPath, virStrerror(errno, ebuf, sizeof ebuf));
+ goto error;
+ }
/* Configuration paths are either ~/.libvirt/qemu/... (session) or
* /etc/libvirt/qemu/... (system).
@@ -1971,6 +2002,10 @@ qemudStartup(int privileged) {
qemudAutostartConfigs(qemu_driver);
+ qemu_driver->workerPool = virThreadPoolNew(0, 1, processWatchdogEvent,
qemu_driver);
+ if (!qemu_driver->workerPool)
+ goto error;
+
if (conn)
virConnectClose(conn);
@@ -2077,6 +2112,7 @@ qemudShutdown(void) {
VIR_FREE(qemu_driver->cacheDir);
VIR_FREE(qemu_driver->saveDir);
VIR_FREE(qemu_driver->snapshotDir);
+ VIR_FREE(qemu_driver->autoDumpPath);
VIR_FREE(qemu_driver->vncTLSx509certdir);
VIR_FREE(qemu_driver->vncListen);
VIR_FREE(qemu_driver->vncPassword);
@@ -2112,6 +2148,7 @@ qemudShutdown(void) {
qemuDriverUnlock(qemu_driver);
virMutexDestroy(&qemu_driver->lock);
+ virThreadPoolFree(qemu_driver->workerPool);
VIR_FREE(qemu_driver);
return 0;
@@ -6263,6 +6300,65 @@ cleanup:
return ret;
}
+static void processWatchdogEvent(void *data, void *opaque)
+{
+ int ret;
+ struct watchdogEvent *wdEvent = data;
+ struct qemud_driver *driver = opaque;
+
+ switch (wdEvent->action) {
+ case VIR_DOMAIN_WATCHDOG_ACTION_DUMP:
+ {
+ char *dumpfile;
+ int i;
+
+ qemuDomainObjPrivatePtr priv = wdEvent->vm->privateData;
+
+ i = virAsprintf(&dumpfile, "%s/%s-%u",
+ driver->autoDumpPath,
+ wdEvent->vm->def->name,
+ (unsigned int)time(NULL));
+
+ qemuDriverLock(driver);
+ virDomainObjLock(wdEvent->vm);
+
+ if (qemuDomainObjBeginJobWithDriver(driver, wdEvent->vm) < 0)
+ break;
+
+ if (!virDomainObjIsActive(wdEvent->vm)) {
+ qemuReportError(VIR_ERR_OPERATION_INVALID,
+ "%s", _("domain is not running"));
+ break;
+ }
+
+ ret = doCoreDump(driver,
+ wdEvent->vm,
+ dumpfile,
+ getCompressionType(driver));
+ if (ret < 0)
+ qemuReportError(VIR_ERR_OPERATION_FAILED,
+ "%s", _("Dump failed"));
+
+ qemuDomainObjEnterMonitorWithDriver(driver, wdEvent->vm);
+ ret = qemuMonitorStartCPUs(priv->mon, NULL);
+ qemuDomainObjExitMonitorWithDriver(driver, wdEvent->vm);
+
+ if (ret < 0)
+ qemuReportError(VIR_ERR_OPERATION_FAILED,
+ "%s", _("Resuming after dump
failed"));
+
+ if (qemuDomainObjEndJob(wdEvent->vm) > 0)
+ virDomainObjUnlock(wdEvent->vm);
+
+ qemuDriverUnlock(driver);
+
+ VIR_FREE(dumpfile);
+ }
+ break;
+ }
+
+ VIR_FREE(wdEvent);
+}
static int qemudDomainHotplugVcpus(virDomainObjPtr vm, unsigned int nvcpus)
{
diff --git a/src/qemu/test_libvirtd_qemu.aug b/src/qemu/test_libvirtd_qemu.aug
index d3ae58d..ce5e69b 100644
--- a/src/qemu/test_libvirtd_qemu.aug
+++ b/src/qemu/test_libvirtd_qemu.aug
@@ -96,6 +96,8 @@ save_image_format = \"gzip\"
dump_image_format = \"gzip\"
+auto_dump_path = \"/var/lib/libvirt/qemu/dump\"
+
hugetlbfs_mount = \"/dev/hugepages\"
set_process_name = 1
--
1.7.3
--
Thanks,
Hu Tao
11:40 p.m.
New subject: [libvirt] [PATCH v5 4/4] Using threadpool API to manage qemud worker
---
daemon/libvirtd.c | 172 +++++++++--------------------------------------------
daemon/libvirtd.h | 2 +
2 files changed, 31 insertions(+), 143 deletions(-)
diff --git a/daemon/libvirtd.c b/daemon/libvirtd.c
index 791b3dc..effa45f 100644
--- a/daemon/libvirtd.c
+++ b/daemon/libvirtd.c
@@ -67,6 +67,7 @@
#include "stream.h"
#include "hooks.h"
#include "virtaudit.h"
+#include "threadpool.h"
#ifdef HAVE_AVAHI
# include "mdns.h"
#endif
@@ -248,7 +249,6 @@ static void sig_handler(int sig, siginfo_t * siginfo,
static void qemudDispatchClientEvent(int watch, int fd, int events, void *opaque);
static void qemudDispatchServerEvent(int watch, int fd, int events, void *opaque);
-static int qemudStartWorker(struct qemud_server *server, struct qemud_worker *worker);
void
qemudClientMessageQueuePush(struct qemud_client_message **queue,
@@ -1458,19 +1458,6 @@ static int qemudDispatchServer(struct qemud_server *server, struct
qemud_socket
server->clients[server->nclients++] = client;
- if (server->nclients > server->nactiveworkers &&
- server->nactiveworkers < server->nworkers) {
- for (i = 0 ; i < server->nworkers ; i++) {
- if (!server->workers[i].hasThread) {
- if (qemudStartWorker(server, &server->workers[i]) < 0)
- return -1;
- server->nactiveworkers++;
- break;
- }
- }
- }
-
-
return 0;
error:
@@ -1534,100 +1521,28 @@ void qemudDispatchClientFailure(struct qemud_client *client) {
VIR_FREE(client->addrstr);
}
-
-/* Caller must hold server lock */
-static struct qemud_client *qemudPendingJob(struct qemud_server *server)
-{
- int i;
- for (i = 0 ; i < server->nclients ; i++) {
- virMutexLock(&server->clients[i]->lock);
- if (server->clients[i]->dx) {
- /* Delibrately don't unlock client - caller wants the lock */
- return server->clients[i];
- }
- virMutexUnlock(&server->clients[i]->lock);
- }
- return NULL;
-}
-
-static void *qemudWorker(void *data)
+static void qemudWorker(void *data, void *opaque)
{
- struct qemud_worker *worker = data;
- struct qemud_server *server = worker->server;
-
- while (1) {
- struct qemud_client *client = NULL;
- struct qemud_client_message *msg;
-
- virMutexLock(&server->lock);
- while ((client = qemudPendingJob(server)) == NULL) {
- if (worker->quitRequest ||
- virCondWait(&server->job, &server->lock) < 0) {
- virMutexUnlock(&server->lock);
- return NULL;
- }
- }
- if (worker->quitRequest) {
- virMutexUnlock(&client->lock);
- virMutexUnlock(&server->lock);
- return NULL;
- }
- worker->processingCall = 1;
- virMutexUnlock(&server->lock);
-
- /* We own a locked client now... */
- client->refs++;
-
- /* Remove our message from dispatch queue while we use it */
- msg = qemudClientMessageQueueServe(&client->dx);
-
- /* This function drops the lock during dispatch,
- * and re-acquires it before returning */
- if (remoteDispatchClientRequest (server, client, msg) < 0) {
- VIR_FREE(msg);
- qemudDispatchClientFailure(client);
- client->refs--;
- virMutexUnlock(&client->lock);
- continue;
- }
-
- client->refs--;
- virMutexUnlock(&client->lock);
-
- virMutexLock(&server->lock);
- worker->processingCall = 0;
- virMutexUnlock(&server->lock);
- }
-}
-
-static int qemudStartWorker(struct qemud_server *server,
- struct qemud_worker *worker) {
- pthread_attr_t attr;
- pthread_attr_init(&attr);
- /* We want to join workers, so don't detach them */
- /*pthread_attr_setdetachstate(&attr, 1);*/
+ struct qemud_server *server = opaque;
+ struct qemud_client *client = data;
+ struct qemud_client_message *msg;
- if (worker->hasThread)
- return -1;
+ virMutexLock(&client->lock);
- worker->server = server;
- worker->hasThread = 1;
- worker->quitRequest = 0;
- worker->processingCall = 0;
+ /* Remove our message from dispatch queue while we use it */
+ msg = qemudClientMessageQueueServe(&client->dx);
- if (pthread_create(&worker->thread,
- &attr,
- qemudWorker,
- worker) != 0) {
- worker->hasThread = 0;
- worker->server = NULL;
- return -1;
+ /* This function drops the lock during dispatch,
+ * and re-acquires it before returning */
+ if (remoteDispatchClientRequest (server, client, msg) < 0) {
+ VIR_FREE(msg);
+ qemudDispatchClientFailure(client);
}
- return 0;
+ client->refs--;
+ virMutexUnlock(&client->lock);
}
-
/*
* Read data into buffer using wire decoding (plain or TLS)
*
@@ -1857,8 +1772,11 @@ readmore:
}
/* Move completed message to the end of the dispatch queue */
- if (msg)
+ if (msg) {
+ client->refs++;
qemudClientMessageQueuePush(&client->dx, msg);
+ ignore_value(virThreadPoolSendJob(server->workerPool, client));
+ }
client->nrequests++;
/* Possibly need to create another receive buffer */
@@ -1870,9 +1788,6 @@ readmore:
client->rx->bufferLength = REMOTE_MESSAGE_HEADER_XDR_LEN;
qemudUpdateClientEvent(client);
-
- /* Tell one of the workers to get on with it... */
- virCondSignal(&server->job);
}
}
}
@@ -2311,10 +2226,14 @@ static void *qemudRunLoop(void *opaque) {
return NULL;
}
- for (i = 0 ; i < min_workers ; i++) {
- if (qemudStartWorker(server, &server->workers[i]) < 0)
- goto cleanup;
- server->nactiveworkers++;
+ server->workerPool = virThreadPoolNew(min_workers,
+ max_workers,
+ qemudWorker,
+ server);
+ if (!server->workerPool) {
+ VIR_ERROR0(_("Failed to create thread pool"));
+ virMutexUnlock(&server->lock);
+ return NULL;
}
for (;!server->quitEventThread;) {
@@ -2367,47 +2286,14 @@ static void *qemudRunLoop(void *opaque) {
goto reprocess;
}
}
-
- /* If number of active workers exceeds both the min_workers
- * threshold and the number of clients, then kill some
- * off */
- for (i = 0 ; (i < server->nworkers &&
- server->nactiveworkers > server->nclients &&
- server->nactiveworkers > min_workers) ; i++) {
-
- if (server->workers[i].hasThread &&
- !server->workers[i].processingCall) {
- server->workers[i].quitRequest = 1;
-
- virCondBroadcast(&server->job);
- virMutexUnlock(&server->lock);
- pthread_join(server->workers[i].thread, NULL);
- virMutexLock(&server->lock);
- server->workers[i].hasThread = 0;
- server->nactiveworkers--;
- }
- }
- }
-
-cleanup:
- for (i = 0 ; i < server->nworkers ; i++) {
- if (!server->workers[i].hasThread)
- continue;
-
- server->workers[i].quitRequest = 1;
- virCondBroadcast(&server->job);
-
- virMutexUnlock(&server->lock);
- pthread_join(server->workers[i].thread, NULL);
- virMutexLock(&server->lock);
- server->workers[i].hasThread = 0;
}
- VIR_FREE(server->workers);
for (i = 0; i < server->nclients; i++)
qemudFreeClient(server->clients[i]);
server->nclients = 0;
VIR_SHRINK_N(server->clients, server->nclients_max, server->nclients_max);
+ virThreadPoolFree(server->workerPool);
+ server->workerPool = NULL;
virMutexUnlock(&server->lock);
return NULL;
}
diff --git a/daemon/libvirtd.h b/daemon/libvirtd.h
index af20e56..dfd54cd 100644
--- a/daemon/libvirtd.h
+++ b/daemon/libvirtd.h
@@ -49,6 +49,7 @@
# include "logging.h"
# include "threads.h"
# include "network.h"
+# include "threadpool.h"
# if WITH_DTRACE
# ifndef LIBVIRTD_PROBES_H
@@ -283,6 +284,7 @@ struct qemud_server {
int privileged;
+ virThreadPoolPtr workerPool;
size_t nworkers;
size_t nactiveworkers;
struct qemud_worker *workers;
--
1.7.3
--
Thanks,
Hu Tao
5103
days inactive
5103
days old
4 comments
1 participants
participants (1)
-
Hu Tao