[libvirt] [PATCH v5 4/7] fdstream: Emit stream abort callback even if poll() doesnt.

[Peter Krempa]

This patch causes the fdstream driver to call the stream event callback
if virStreamAbort() is called on a stream using this driver.

A remote handler for a stream can only detect changes via stream events,
so this event callback is necessary in order to enable a daemon to abort
a stream in such a way that the client will see the change.

* src/fdstream.c:
        - modify close function to call stream event callback
---
Diff to v4:
- reworded commit message
- fixed year
- added caching of pointers to callback before dropping the lock
- moved lines belonging to this patch here

Note to comment in v4 review:
fdst->abortCallbackCalled needs to be cleared only if the callback handler
is set. This is not needed when we just update the events for the callback.

 src/fdstream.c |   63 ++++++++++++++++++++++++++++++++++++++++++++++++++-----
 1 files changed, 57 insertions(+), 6 deletions(-)

diff --git a/src/fdstream.c b/src/fdstream.c
index 841f979..39a8753 100644
--- a/src/fdstream.c
+++ b/src/fdstream.c
@@ -1,7 +1,7 @@
 /*
- * fdstream.h: generic streams impl for file descriptors
+ * fdstream.c: generic streams impl for file descriptors
  *
- * Copyright (C) 2009-2011 Red Hat, Inc.
+ * Copyright (C) 2009-2012 Red Hat, Inc.
  *
  * This library is free software; you can redistribute it and/or
  * modify it under the terms of the GNU Lesser General Public
@@ -55,6 +55,7 @@ struct virFDStreamData {
     unsigned long long length;

     int watch;
+    int events;         /* events the stream callback is subscribed for */
     bool cbRemoved;
     bool dispatching;
     bool closed;
@@ -62,6 +63,10 @@ struct virFDStreamData {
     void *opaque;
     virFreeCallback ff;

+    /* don't call the abort callback more than once */
+    bool abortCallbackCalled;
+    bool abortCallbackDispatching;
+
     virMutex lock;
 };

@@ -92,6 +97,7 @@ static int virFDStreamRemoveCallback(virStreamPtr stream)
     fdst->watch = 0;
     fdst->ff = NULL;
     fdst->cb = NULL;
+    fdst->events = 0;
     fdst->opaque = NULL;

     ret = 0;
@@ -120,6 +126,7 @@ static int virFDStreamUpdateCallback(virStreamPtr stream, int events)
     }

     virEventUpdateHandle(fdst->watch, events);
+    fdst->events = events;

     ret = 0;

@@ -214,6 +221,8 @@ virFDStreamAddCallback(virStreamPtr st,
     fdst->cb = cb;
     fdst->opaque = opaque;
     fdst->ff = ff;
+    fdst->events = events;
+    fdst->abortCallbackCalled = false;
     virStreamRef(st);

     ret = 0;
@@ -225,18 +234,48 @@ cleanup:


 static int
-virFDStreamClose(virStreamPtr st)
+virFDStreamCloseInt(virStreamPtr st, bool streamAbort)
 {
-    struct virFDStreamData *fdst = st->privateData;
+    struct virFDStreamData *fdst;
+    virStreamEventCallback cb;
+    void *opaque;
     int ret;

     VIR_DEBUG("st=%p", st);

-    if (!fdst)
+    if (!st || !(fdst = st->privateData) || fdst->abortCallbackDispatching)
         return 0;

     virMutexLock(&fdst->lock);

+    /* aborting the stream, ensure the callback is called if it's
+     * registered for stream error event */
+    if (streamAbort &&
+        fdst->cb &&
+        (fdst->events & (VIR_STREAM_EVENT_READABLE |
+                         VIR_STREAM_EVENT_WRITABLE))) {
+        /* don't enter this function accidentally from the callback again */
+        if (fdst->abortCallbackCalled) {
+            virMutexUnlock(&fdst->lock);
+            return 0;
+        }
+
+        fdst->abortCallbackCalled = true;
+        fdst->abortCallbackDispatching = true;
+
+        /* cache the pointers */
+        cb = fdst->cb;
+        opaque = fdst->opaque;
+        virMutexUnlock(&fdst->lock);
+
+        /* call failure callback, poll reports nothing on closed fd */
+        (cb)(st, VIR_STREAM_EVENT_ERROR, opaque);
+
+        virMutexLock(&fdst->lock);
+        fdst->abortCallbackDispatching = false;
+    }
+
+    /* mutex locked */
     ret = VIR_CLOSE(fdst->fd);
     if (fdst->cmd) {
         char buf[1024];
@@ -286,6 +325,18 @@ virFDStreamClose(virStreamPtr st)
     return ret;
 }

+static int
+virFDStreamClose(virStreamPtr st)
+{
+    return virFDStreamCloseInt(st, false);
+}
+
+static int
+virFDStreamAbort(virStreamPtr st)
+{
+    return virFDStreamCloseInt(st, true);
+}
+
 static int virFDStreamWrite(virStreamPtr st, const char *bytes, size_t nbytes)
 {
     struct virFDStreamData *fdst = st->privateData;
@@ -392,7 +443,7 @@ static virStreamDriver virFDStreamDrv = {
     .streamSend = virFDStreamWrite,
     .streamRecv = virFDStreamRead,
     .streamFinish = virFDStreamClose,
-    .streamAbort = virFDStreamClose,
+    .streamAbort = virFDStreamAbort,
     .streamAddCallback = virFDStreamAddCallback,
     .streamUpdateCallback = virFDStreamUpdateCallback,
     .streamRemoveCallback = virFDStreamRemoveCallback
-- 
1.7.3.4