Re: [libvirt] Re: [PATCH 07/12] Domain Events - remote driver

Deliver local callbacks in response to remote events remote_internal.c | 255 ++++++++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 248 insertions(+), 7 deletions(-)

@@ -680,6 +689,26 @@ doRemoteOpen (virConnectPtr conn, (xdrproc_t) xdr_void, (char *) NULL) == -1) goto failed; + if(VIR_ALLOC(priv->domainEvents)<0) { + error(conn, VIR_ERR_INVALID_ARG, _("Error allocating domainEvents")); + goto failed; + } + + DEBUG0("Adding Handler for remote events"); + /* Set up a callback to listen on the socket data */ + if (virEventAddHandle(priv->sock, + POLLIN | POLLERR | POLLHUP, + remoteDomainEventFired, + conn) < 0) { + DEBUG0("virEventAddHandle failed: No addHandleImpl defined. continuing without events."); + } + + DEBUG0("Adding Timeout for remote event queue flushing"); + if ( (priv->eventFlushTimer = virEventAddTimeout(0, + remoteDomainEventQueueFlush, + conn)) < 0) {

+ +static int remoteDomainEventRegister (virConnectPtr conn, + void *callback, + void *opaque) +{ + struct private_data *priv = conn->privateData; + + /* dispatch an rpc - so the server sde can track + how many callbacks are regstered */ + remote_domain_events_register_args args; + args.callback = (unsigned long)callback; + args.user_data = (unsigned long)opaque;

/*----------------------------------------------------------------------*/ @@ -4367,6 +4444,7 @@ call (virConnectPtr conn, struct private_data *priv, really_write (conn, priv, flags & REMOTE_CALL_IN_OPEN, buffer, len-4) == -1) return -1; +retry_read: /* Read and deserialise length word. */ if (really_read (conn, priv, flags & REMOTE_CALL_IN_OPEN, buffer2, sizeof buffer2) == -1) return -1; @@ -4418,10 +4496,19 @@ call (virConnectPtr conn, struct private_data *priv, return -1; } - /* If we extend the server to actually send asynchronous messages, then - * we'll need to change this so that it can recognise an asynch - * message being received at this point. - */ + if (hdr.proc == REMOTE_PROC_DOMAIN_EVENT && + hdr.direction == REMOTE_MESSAGE) { + /* An async message has come in while we were waiting for the + * response. Process it to pull it off the wire, and try again + */ + DEBUG0("Encountered an event while waiting for a response"); + + remoteDomainQueueEvent(conn, &xdr);

+/** + * remoteDomainReadEvent + * + * Read the event data off the wire + */ +static int remoteDomainReadEvent(virConnectPtr conn, XDR *xdr, + virDomainPtr *dom, int *event, + virConnectDomainEventCallback *cb, + void **opaque) +{ + remote_domain_event_ret ret; + memset (&ret, 0, sizeof ret); + + /* unmarshall parameters, and process it*/ + if (! xdr_remote_domain_event_ret(xdr, &ret) ) { + error (conn, VIR_ERR_RPC, _("remoteDomainProcessEvent: unmarshalling ret")); + return -1; + } + + *dom = get_nonnull_domain(conn,ret.dom); + *event = ret.event; + *cb = (virConnectDomainEventCallback)ret.callback; + *opaque = (void *)ret.user_data; + + return 0; +} + +static void remoteDomainProcessEvent(virConnectPtr conn, XDR *xdr) +{ + virDomainPtr dom; + int event; + virConnectDomainEventCallback cb; + void *opaque; + if(!remoteDomainReadEvent(conn, xdr, &dom, &event, &cb, &opaque)) { + DEBUG0("Calling domain event callback (no queue)"); + if(cb) + cb(conn,dom,event,opaque);

+ } +} + +static void remoteDomainQueueEvent(virConnectPtr conn, XDR *xdr) +{ + virDomainPtr dom; + int event; + virConnectDomainEventCallback cb; + void *opaque; + struct private_data *priv = conn->privateData; + + if(!remoteDomainReadEvent(conn, xdr, &dom, &event, &cb, &opaque)) + { + if( virDomainEventCallbackQueuePush(priv->domainEvents, dom, event, cb, opaque) < 0 ) { + DEBUG("%s", "Error adding event to queue"); + } + } +} + +/** remoteDomainEventFired: + * + * The callback for monitoring the remote socket + * for event data + */ +void remoteDomainEventFired(int fd ATTRIBUTE_UNUSED, + int event ATTRIBUTE_UNUSED, + void *opaque) +{ + char buffer[REMOTE_MESSAGE_MAX]; + char buffer2[4]; + struct remote_message_header hdr; + XDR xdr; + int len; + + virConnectPtr conn = opaque; + struct private_data *priv = conn->privateData; + + DEBUG("%s : Event fired", __FUNCTION__); + + /* Read and deserialise length word. */ + if (really_read (conn, priv, 0, buffer2, sizeof buffer2) == -1) + return; + + xdrmem_create (&xdr, buffer2, sizeof buffer2, XDR_DECODE); + if (!xdr_int (&xdr, &len)) { + error (conn, VIR_ERR_RPC, _("xdr_int (length word, reply)")); + return; + } + xdr_destroy (&xdr); + + /* Length includes length word - adjust to real length to read. */ + len -= 4; + + if (len < 0 || len > REMOTE_MESSAGE_MAX) { + error (conn, VIR_ERR_RPC, _("packet received from server too large")); + return; + } + + /* Read reply header and what follows (either a ret or an error). */ + if (really_read (conn, priv, 0, buffer, len) == -1) { + error (conn, VIR_ERR_RPC, _("error reading buffer from memory")); + return; + } + + /* Deserialise reply header. */ + xdrmem_create (&xdr, buffer, len, XDR_DECODE); + if (!xdr_remote_message_header (&xdr, &hdr)) { + error (conn, VIR_ERR_RPC, _("invalid header in event firing")); + return; + } + + if (hdr.proc == REMOTE_PROC_DOMAIN_EVENT && + hdr.direction == REMOTE_MESSAGE) { + DEBUG0("Encountered an async event"); + remoteDomainProcessEvent(conn, &xdr); + } else { + DEBUG0("invalid proc in event firing"); + error (conn, VIR_ERR_RPC, _("invalid proc in event firing")); + } +} + +void remoteDomainEventQueueFlush(int timer ATTRIBUTE_UNUSED, + void *opaque) +{ + virDomainEventPtr domEvent; + virConnectPtr conn = opaque; + struct private_data *priv = conn->privateData; + + DEBUG0("Flushing domain events"); + while( (domEvent = virDomainEventCallbackQueuePop(priv->domainEvents)) ) { + DEBUG(" Flushing %p", domEvent); + if(domEvent->cb) + domEvent->cb(domEvent->dom->conn, + domEvent->dom, + domEvent->event, + domEvent->opaque);

+ VIR_FREE(domEvent); + }

+}