NuKVM/libvirt
3
0
Fork 0
mirror of https://github.com/libvirt/libvirt.git synced 2025-02-25 18:55:26 -06:00
Code Issues Packages Projects Releases Wiki Activity

qemu: convert agent to use the per-VM event loop

Browse Source 
This converts the QEMU agent APIs to use the per-VM
event loop, which involves switching from virEvent APIs
to GMainContext / GSource APIs.

A GSocket is used as a convenient way to create a GSource
for a socket, but is not yet used for actual I/O.

Reviewed-by: Michal Privoznik <mprivozn@redhat.com>
Signed-off-by: Daniel P. Berrangé <berrange@redhat.com>
This commit is contained in:
Daniel P. Berrangé 2020-02-12 14:54:19 +00:00
parent 0d62faf62a
commit a18f2c52ac
4 changed files with 84 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

146
src/qemu/qemu_agent.c
View File

@ -25,6 +25,7 @@
#include <unistd.h> #include <unistd.h>
#include <fcntl.h> #include <fcntl.h>
#include <sys/time.h> #include <sys/time.h>
#include <gio/gio.h>
#include "qemu_agent.h" #include "qemu_agent.h"
#include "qemu_domain.h" #include "qemu_domain.h"
@ -101,7 +102,10 @@ struct _qemuAgent {
virCond notify; virCond notify;
int fd; int fd;
int watch;
GMainContext *context;
GSocket *socket;
GSource *watch;
bool running; bool running;
@ -172,6 +176,7 @@ static void qemuAgentDispose(void *obj)
(agent->cb->destroy)(agent, agent->vm); (agent->cb->destroy)(agent, agent->vm);
virCondDestroy(&agent->notify); virCondDestroy(&agent->notify);
VIR_FREE(agent->buffer); VIR_FREE(agent->buffer);
g_main_context_unref(agent->context);
virResetError(&agent->lastError); virResetError(&agent->lastError);
} }
@ -188,13 +193,6 @@ qemuAgentOpenUnix(const char *socketpath)
return -1; return -1;
} }
if (virSetNonBlock(agentfd) < 0) {
virReportSystemError(errno, "%s",
_("Unable to put monitor "
"into non-blocking mode"));
goto error;
}
if (virSetCloseExec(agentfd) < 0) { if (virSetCloseExec(agentfd) < 0) {
virReportSystemError(errno, "%s", virReportSystemError(errno, "%s",
_("Unable to set agent " _("Unable to set agent "
@ -498,28 +496,62 @@ qemuAgentIORead(qemuAgentPtr agent)
} }
static void qemuAgentUpdateWatch(qemuAgentPtr agent) static gboolean
{ qemuAgentIO(GSocket *socket,
int events = GIOCondition cond,
VIR_EVENT_HANDLE_HANGUP | gpointer opaque);
VIR_EVENT_HANDLE_ERROR;
if (!agent->watch)
return; static void
qemuAgentRegister(qemuAgentPtr agent)
{
GIOCondition cond = 0;
if (agent->lastError.code == VIR_ERR_OK) { if (agent->lastError.code == VIR_ERR_OK) {
events |= VIR_EVENT_HANDLE_READABLE; cond |= G_IO_IN;
if (agent->msg && agent->msg->txOffset < agent->msg->txLength) if (agent->msg && agent->msg->txOffset < agent->msg->txLength)
events |= VIR_EVENT_HANDLE_WRITABLE; cond |= G_IO_OUT;
} }
virEventUpdateHandle(agent->watch, events); agent->watch = g_socket_create_source(agent->socket,
cond,
NULL);
virObjectRef(agent);
g_source_set_callback(agent->watch,
(GSourceFunc)qemuAgentIO,
agent,
(GDestroyNotify)virObjectUnref);
g_source_attach(agent->watch,
agent->context);
} }
static void static void
qemuAgentIO(int watch, int fd, int events, void *opaque) qemuAgentUnregister(qemuAgentPtr agent)
{
if (agent->watch) {
g_source_destroy(agent->watch);
g_source_unref(agent->watch);
agent->watch = NULL;
}
}
static void qemuAgentUpdateWatch(qemuAgentPtr agent)
{
qemuAgentUnregister(agent);
if (agent->socket)
qemuAgentRegister(agent);
}
static gboolean
qemuAgentIO(GSocket *socket G_GNUC_UNUSED,
GIOCondition cond,
gpointer opaque)
{ {
qemuAgentPtr agent = opaque; qemuAgentPtr agent = opaque;
bool error = false; bool error = false;
@ -529,45 +561,36 @@ qemuAgentIO(int watch, int fd, int events, void *opaque)
/* lock access to the agent and protect fd */ /* lock access to the agent and protect fd */
virObjectLock(agent); virObjectLock(agent);
#if DEBUG_IO #if DEBUG_IO
VIR_DEBUG("Agent %p I/O on watch %d fd %d events %d", agent, watch, fd, events); VIR_DEBUG("Agent %p I/O on watch %d socket %p cond %d", agent, agent->socket, cond);
#endif #endif
if (agent->fd == -1 || agent->watch == 0) { if (agent->fd == -1 || !agent->watch) {
virObjectUnlock(agent); virObjectUnlock(agent);
virObjectUnref(agent); virObjectUnref(agent);
return; return G_SOURCE_REMOVE;
} }
if (agent->fd != fd || agent->watch != watch) { if (agent->lastError.code != VIR_ERR_OK) {
if (events & (VIR_EVENT_HANDLE_HANGUP | VIR_EVENT_HANDLE_ERROR)) if (cond & (G_IO_HUP | G_IO_ERR))
eof = true;
virReportError(VIR_ERR_INTERNAL_ERROR,
_("event from unexpected fd %d!=%d / watch %d!=%d"),
agent->fd, fd, agent->watch, watch);
error = true;
} else if (agent->lastError.code != VIR_ERR_OK) {
if (events & (VIR_EVENT_HANDLE_HANGUP | VIR_EVENT_HANDLE_ERROR))
eof = true; eof = true;
error = true; error = true;
} else { } else {
if (events & VIR_EVENT_HANDLE_WRITABLE) { if (cond & G_IO_OUT) {
if (qemuAgentIOWrite(agent) < 0) if (qemuAgentIOWrite(agent) < 0)
error = true; error = true;
events &= ~VIR_EVENT_HANDLE_WRITABLE;
} }
if (!error && if (!error &&
events & VIR_EVENT_HANDLE_READABLE) { cond & G_IO_IN) {
int got = qemuAgentIORead(agent); int got = qemuAgentIORead(agent);
events &= ~VIR_EVENT_HANDLE_READABLE;
if (got < 0) { if (got < 0) {
error = true; error = true;
} else if (got == 0) { } else if (got == 0) {
eof = true; eof = true;
} else { } else {
/* Ignore hangup/error events if we read some data, to /* Ignore hangup/error cond if we read some data, to
* give time for that data to be consumed */ * give time for that data to be consumed */
events = 0; cond = 0;
if (qemuAgentIOProcess(agent) < 0) if (qemuAgentIOProcess(agent) < 0)
error = true; error = true;
@ -575,25 +598,17 @@ qemuAgentIO(int watch, int fd, int events, void *opaque)
} }
if (!error && if (!error &&
events & VIR_EVENT_HANDLE_HANGUP) { cond & G_IO_HUP) {
virReportError(VIR_ERR_INTERNAL_ERROR, "%s", virReportError(VIR_ERR_INTERNAL_ERROR, "%s",
_("End of file from agent socket")); _("End of file from agent socket"));
eof = true; eof = true;
events &= ~VIR_EVENT_HANDLE_HANGUP;
} }
if (!error && !eof && if (!error && !eof &&
events & VIR_EVENT_HANDLE_ERROR) { cond & G_IO_ERR) {
virReportError(VIR_ERR_INTERNAL_ERROR, "%s", virReportError(VIR_ERR_INTERNAL_ERROR, "%s",
_("Invalid file descriptor while waiting for agent")); _("Invalid file descriptor while waiting for agent"));
eof = true; eof = true;
events &= ~VIR_EVENT_HANDLE_ERROR;
}
if (!error && events) {
virReportError(VIR_ERR_INTERNAL_ERROR,
_("Unhandled event %d for agent fd %d"),
events, agent->fd);
error = true;
} }
} }
@ -649,15 +664,19 @@ qemuAgentIO(int watch, int fd, int events, void *opaque)
virObjectUnlock(agent); virObjectUnlock(agent);
virObjectUnref(agent); virObjectUnref(agent);
} }
return G_SOURCE_REMOVE;
} }
qemuAgentPtr qemuAgentPtr
qemuAgentOpen(virDomainObjPtr vm, qemuAgentOpen(virDomainObjPtr vm,
const virDomainChrSourceDef *config, const virDomainChrSourceDef *config,
GMainContext *context,
qemuAgentCallbacksPtr cb) qemuAgentCallbacksPtr cb)
{ {
qemuAgentPtr agent; qemuAgentPtr agent;
g_autoptr(GError) gerr = NULL;
if (!cb || !cb->eofNotify) { if (!cb || !cb->eofNotify) {
virReportError(VIR_ERR_INTERNAL_ERROR, "%s", virReportError(VIR_ERR_INTERNAL_ERROR, "%s",
@ -693,22 +712,20 @@ qemuAgentOpen(virDomainObjPtr vm,
if (agent->fd == -1) if (agent->fd == -1)
goto cleanup; goto cleanup;
virObjectRef(agent); agent->context = g_main_context_ref(context);
if ((agent->watch = virEventAddHandle(agent->fd,
VIR_EVENT_HANDLE_HANGUP | agent->socket = g_socket_new_from_fd(agent->fd, &gerr);
VIR_EVENT_HANDLE_ERROR | if (!agent->socket) {
VIR_EVENT_HANDLE_READABLE, virReportError(VIR_ERR_INTERNAL_ERROR,
qemuAgentIO, _("Unable to create socket object: %s"),
agent, gerr->message);
virObjectFreeCallback)) < 0) {
virObjectUnref(agent);
virReportError(VIR_ERR_INTERNAL_ERROR, "%s",
_("unable to register agent events"));
goto cleanup; goto cleanup;
} }
qemuAgentRegister(agent);
agent->running = true; agent->running = true;
VIR_DEBUG("New agent %p fd =%d watch=%d", agent, agent->fd, agent->watch); VIR_DEBUG("New agent %p fd=%d", agent, agent->fd);
return agent; return agent;
@ -763,12 +780,11 @@ void qemuAgentClose(qemuAgentPtr agent)
virObjectLock(agent); virObjectLock(agent);
if (agent->fd >= 0) { if (agent->socket) {
if (agent->watch) { qemuAgentUnregister(agent);
virEventRemoveHandle(agent->watch); g_object_unref(agent->socket);
agent->watch = 0; agent->socket = NULL;
} agent->fd = -1;
VIR_FORCE_CLOSE(agent->fd);
} }
qemuAgentNotifyCloseLocked(agent); qemuAgentNotifyCloseLocked(agent);

1
src/qemu/qemu_agent.h
View File

@ -41,6 +41,7 @@ struct _qemuAgentCallbacks {
qemuAgentPtr qemuAgentOpen(virDomainObjPtr vm, qemuAgentPtr qemuAgentOpen(virDomainObjPtr vm,
const virDomainChrSourceDef *config, const virDomainChrSourceDef *config,
GMainContext *context,
qemuAgentCallbacksPtr cb); qemuAgentCallbacksPtr cb);
void qemuAgentClose(qemuAgentPtr mon); void qemuAgentClose(qemuAgentPtr mon);

1
src/qemu/qemu_process.c
View File

@ -237,6 +237,7 @@ qemuConnectAgent(virQEMUDriverPtr driver, virDomainObjPtr vm)
agent = qemuAgentOpen(vm, agent = qemuAgentOpen(vm,
config->source, config->source,
virEventThreadGetContext(priv->eventThread),
&agentCallbacks); &agentCallbacks);
virObjectLock(vm); virObjectLock(vm);

1
tests/qemumonitortestutils.c
View File

@ -1406,6 +1406,7 @@ qemuMonitorTestNewAgent(virDomainXMLOptionPtr xmlopt)
if (!(test->agent = qemuAgentOpen(test->vm, if (!(test->agent = qemuAgentOpen(test->vm,
&src, &src,
virEventThreadGetContext(test->eventThread),
&qemuMonitorTestAgentCallbacks))) &qemuMonitorTestAgentCallbacks)))
goto error; goto error;