events: Propose a separate lock for event queue

Currently, push & pop from event queue (both server & client side)
rely on lock from higher levels, e.g. on driver lock (qemu),
private_data (remote), ...; This alone is not sufficient as not
every function that interacts with this queue can/does lock,
esp. in client where we have a different approach, "passing
the buck".

Therefore we need a separate lock just to protect event queue.

For more info see:
https://bugzilla.redhat.com/show_bug.cgi?id=743817

src/conf/domain_event.c

@@ -547,6 +547,18 @@ virDomainEventQueuePtr virDomainEventQueueNew(void)
     return ret;
 }
 
+static void
+virDomainEventStateLock(virDomainEventStatePtr state)
+{
+    virMutexLock(&state->lock);
+}
+
+static void
+virDomainEventStateUnlock(virDomainEventStatePtr state)
+{
+    virMutexUnlock(&state->lock);
+}
+
 /**
  * virDomainEventStateFree:
  * @list: virDomainEventStatePtr to free
@@ -564,6 +576,8 @@ virDomainEventStateFree(virDomainEventStatePtr state)
 
     if (state->timer != -1)
         virEventRemoveTimeout(state->timer);
+
+    virMutexDestroy(&state->lock);
     VIR_FREE(state);
 }
 
@@ -590,6 +604,13 @@ virDomainEventStateNew(virEventTimeoutCallback timeout_cb,
         goto error;
     }
 
+    if (virMutexInit(&state->lock) < 0) {
+        virReportSystemError(errno, "%s",
+                             _("unable to initialize state mutex"));
+        VIR_FREE(state);
+        goto error;
+    }
+
     if (VIR_ALLOC(state->callbacks) < 0) {
         virReportOOMError();
         goto error;
@@ -1166,6 +1187,8 @@ virDomainEventStateQueue(virDomainEventStatePtr state,
         return;
     }
 
+    virDomainEventStateLock(state);
+
     if (virDomainEventQueuePush(state->queue, event) < 0) {
         VIR_DEBUG("Error adding event to queue");
         virDomainEventFree(event);
@@ -1173,6 +1196,7 @@ virDomainEventStateQueue(virDomainEventStatePtr state,
 
     if (state->queue->count == 1)
         virEventUpdateTimeout(state->timer, 0);
+    virDomainEventStateUnlock(state);
 }
 
 void
@@ -1182,6 +1206,7 @@ virDomainEventStateFlush(virDomainEventStatePtr state,
 {
     virDomainEventQueue tempQueue;
 
+    virDomainEventStateLock(state);
     state->isDispatching = true;
 
     /* Copy the queue, so we're reentrant safe when dispatchFunc drops the
@@ -1190,17 +1215,20 @@ virDomainEventStateFlush(virDomainEventStatePtr state,
     tempQueue.events = state->queue->events;
     state->queue->count = 0;
     state->queue->events = NULL;
-
     virEventUpdateTimeout(state->timer, -1);
+    virDomainEventStateUnlock(state);
+
     virDomainEventQueueDispatch(&tempQueue,
                                 state->callbacks,
                                 dispatchFunc,
                                 opaque);
 
     /* Purge any deleted callbacks */
+    virDomainEventStateLock(state);
     virDomainEventCallbackListPurgeMarked(state->callbacks);
 
     state->isDispatching = false;
+    virDomainEventStateUnlock(state);
 }
 
 int
@@ -1208,11 +1236,16 @@ virDomainEventStateDeregister(virConnectPtr conn,
                               virDomainEventStatePtr state,
                               virConnectDomainEventCallback callback)
 {
+    int ret;
+
+    virDomainEventStateLock(state);
     if (state->isDispatching)
-        return virDomainEventCallbackListMarkDelete(conn,
+        ret = virDomainEventCallbackListMarkDelete(conn,
-                                                    state->callbacks, callback);
+                                                   state->callbacks, callback);
     else
-        return virDomainEventCallbackListRemove(conn, state->callbacks, callback);
+        ret = virDomainEventCallbackListRemove(conn, state->callbacks, callback);
+    virDomainEventStateUnlock(state);
+    return ret;
 }
 
 int
@@ -1220,10 +1253,15 @@ virDomainEventStateDeregisterAny(virConnectPtr conn,
                                  virDomainEventStatePtr state,
                                  int callbackID)
 {
+    int ret;
+
+    virDomainEventStateLock(state);
     if (state->isDispatching)
-        return virDomainEventCallbackListMarkDeleteID(conn,
+        ret = virDomainEventCallbackListMarkDeleteID(conn,
-                                                      state->callbacks, callbackID);
+                                                     state->callbacks, callbackID);
     else
-        return virDomainEventCallbackListRemoveID(conn,
+        ret = virDomainEventCallbackListRemoveID(conn,
-                                                  state->callbacks, callbackID);
+                                                 state->callbacks, callbackID);
+    virDomainEventStateUnlock(state);
+    return ret;
 }

src/conf/domain_event.h

@@ -61,6 +61,7 @@ struct _virDomainEventState {
     int timer;
     /* Flag if we're in process of dispatching */
     bool isDispatching;
+    virMutex lock;
 };
 typedef struct _virDomainEventState virDomainEventState;
 typedef virDomainEventState *virDomainEventStatePtr;