[RESEND,v2,01/18] net/colo: Add notifier/callback related helpers for filter

Commit Message

Zhanghailiang April 22, 2017, 8:35 a.m. UTC

We will use this notifier to help COLO to notify filter object
to do something, like do checkpoint, or process failover event.

Cc: Jason Wang <[e-mail]>
Signed-off-by: zhanghailiang <[e-mail]>
Signed-off-by: Zhang Chen <[e-mail]>
Signed-off-by: Li Zhijian <[e-mail]>
---
 net/colo.c | 105 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 net/colo.h |  19 +++++++++++
 2 files changed, 124 insertions(+)

Comments

Jason Wang April 25, 2017, 11:40 a.m. UTC | #1

On 2017年04月22日 16:35, zhanghailiang wrote:
> We will use this notifier to help COLO to notify filter object
> to do something, like do checkpoint, or process failover event.
>
> Cc: Jason Wang<[e-mail]>
> Signed-off-by: zhanghailiang<[e-mail]>
> Signed-off-by: Zhang Chen<[e-mail]>
> Signed-off-by: Li Zhijian<[e-mail]>
> ---
>   net/colo.c | 105 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
>   net/colo.h |  19 +++++++++++
>   2 files changed, 124 insertions(+)

Went through this series and I feel that the code duplicates (at least 
part) functionality of IOThread (iothread.c) . So I come to an idea that 
switch to use IOThread for COLO comparing thread then you can use aio bh 
to do the inter process communication.

Thoughts?

Thanks

Zhanghailiang April 26, 2017, 8:14 a.m. UTC | #2

On 2017/4/25 19:40, Jason Wang wrote:
>
> On 2017年04月22日 16:35, zhanghailiang wrote:
>> We will use this notifier to help COLO to notify filter object
>> to do something, like do checkpoint, or process failover event.
>>
>> Cc: Jason Wang<[e-mail]>
>> Signed-off-by: zhanghailiang<[e-mail]>
>> Signed-off-by: Zhang Chen<[e-mail]>
>> Signed-off-by: Li Zhijian<[e-mail]>
>> ---
>>    net/colo.c | 105 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
>>    net/colo.h |  19 +++++++++++
>>    2 files changed, 124 insertions(+)
> Went through this series and I feel that the code duplicates (at least
> part) functionality of IOThread (iothread.c) . So I come to an idea that,
> switch to use IOThread for COLO comparing thread then you can use aio bh
> to do the inter process communication.
>
> Thoughts?

Interesting idea, but after investigation, we found it will make things quite complex
if we use iothread for net compare (Actually, i don't think we can use it directly for COLO),
It seems to be special for block, you can see some codes call block helpers.
Besides, we still have to realize a way to communicate with COLO frame for filters  (I mean patch 16/17/18),
even we use it after reconstruct those iothread part. ;)

Thanks,
Hailiang

> Thanks
>
> .
>

Jason Wang April 26, 2017, 9:14 a.m. UTC | #3

On 2017年04月26日 16:14, Hailiang Zhang wrote:
> On 2017/4/25 19:40, Jason Wang wrote:
>>
>> On 2017年04月22日 16:35, zhanghailiang wrote:
>>> We will use this notifier to help COLO to notify filter object
>>> to do something, like do checkpoint, or process failover event.
>>>
>>> Cc: Jason Wang<[e-mail]>
>>> Signed-off-by: zhanghailiang<[e-mail]>
>>> Signed-off-by: Zhang Chen<[e-mail]>
>>> Signed-off-by: Li Zhijian<[e-mail]>
>>> ---
>>>    net/colo.c | 105 
>>> +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
>>>    net/colo.h |  19 +++++++++++
>>>    2 files changed, 124 insertions(+)
>> Went through this series and I feel that the code duplicates (at least
>> part) functionality of IOThread (iothread.c) . So I come to an idea 
>> that,
>> switch to use IOThread for COLO comparing thread then you can use aio bh
>> to do the inter process communication.
>>
>> Thoughts?
>
> Interesting idea, but after investigation, we found it will make 
> things quite complex
> if we use iothread for net compare (Actually, i don't think we can use 
> it directly for COLO),
> It seems to be special for block, you can see some codes call block 
> helpers.

You mean iothread_stop_all() which was called by main()? It looks just 
use to change aio context back to main context. And I don't think you 
need to care about that.

> Besides, we still have to realize a way to communicate with COLO frame 
> for filters  (I mean patch 16/17/18),
> even we use it after reconstruct those iothread part. ;)
>

I think you can use bh, and e.g go through all the filter in the bh.

Thanks

> Thanks,
> Hailiang
>
>> Thanks
>>
>> .
>>
>
>
>

Patch

diff --git a/net/colo.c b/net/colo.c
index 8cc166b..8aef670 100644
--- a/net/colo.c
+++ b/net/colo.c
@@ -15,6 +15,7 @@ 
 #include "qemu/osdep.h"
 #include "trace.h"
 #include "net/colo.h"
+#include "qapi/error.h"
 
 uint32_t connection_key_hash(const void *opaque)
 {
@@ -209,3 +210,107 @@  Connection *connection_get(GHashTable *connection_track_table,
 
     return conn;
 }
+
+static gboolean
+filter_notify_prepare(GSource *source, gint *timeout)
+{
+    *timeout = -1;
+
+    return FALSE;
+}
+
+static gboolean
+filter_notify_check(GSource *source)
+{
+    FilterNotifier *notify = (FilterNotifier *)source;
+
+    return notify->pfd.revents & (G_IO_IN | G_IO_HUP | G_IO_ERR);
+}
+
+static gboolean
+filter_notify_dispatch(GSource *source,
+                       GSourceFunc callback,
+                       gpointer user_data)
+{
+    FilterNotifier *notify = (FilterNotifier *)source;
+    int revents;
+    uint64_t value;
+    int ret;
+
+    revents = notify->pfd.revents & notify->pfd.events;
+    if (revents & (G_IO_IN | G_IO_HUP | G_IO_ERR)) {
+        ret = filter_notifier_get(notify, &value);
+        if (notify->cb && !ret) {
+            notify->cb(notify, value);
+        }
+    }
+    return TRUE;
+}
+
+static void
+filter_notify_finalize(GSource *source)
+{
+    FilterNotifier *notify = (FilterNotifier *)source;
+
+    event_notifier_cleanup(&notify->event);
+}
+
+static GSourceFuncs notifier_source_funcs = {
+    filter_notify_prepare,
+    filter_notify_check,
+    filter_notify_dispatch,
+    filter_notify_finalize,
+};
+
+FilterNotifier *filter_notifier_new(FilterNotifierCallback *cb,
+                    void *opaque, Error **errp)
+{
+    FilterNotifier *notify;
+    int ret;
+
+    notify = (FilterNotifier *)g_source_new(&notifier_source_funcs,
+                sizeof(FilterNotifier));
+    ret = event_notifier_init(&notify->event, false);
+    if (ret < 0) {
+        error_setg_errno(errp, -ret, "Failed to initialize event notifier");
+        goto fail;
+    }
+    notify->pfd.fd = event_notifier_get_fd(&notify->event);
+    notify->pfd.events = G_IO_IN | G_IO_HUP | G_IO_ERR;
+    notify->cb = cb;
+    notify->opaque = opaque;
+    g_source_add_poll(&notify->source, &notify->pfd);
+
+    return notify;
+
+fail:
+    g_source_destroy(&notify->source);
+    return NULL;
+}
+
+int filter_notifier_set(FilterNotifier *notify, uint64_t value)
+{
+    ssize_t ret;
+
+    do {
+        ret = write(notify->event.wfd, &value, sizeof(value));
+    } while (ret < 0 && errno == EINTR);
+
+    /* EAGAIN is fine, a read must be pending.  */
+    if (ret < 0 && errno != EAGAIN) {
+        return -errno;
+    }
+    return 0;
+}
+
+int filter_notifier_get(FilterNotifier *notify, uint64_t *value)
+{
+    ssize_t len;
+
+    /* Drain the notify pipe.  For eventfd, only 8 bytes will be read.  */
+    do {
+        len = read(notify->event.rfd, value, sizeof(*value));
+    } while ((len == -1 && errno == EINTR));
+
+    return len != sizeof(*value) ? -1 : 0;
+}
diff --git a/net/colo.h b/net/colo.h
index cd9027f..b586db3 100644
--- a/net/colo.h
+++ b/net/colo.h
@@ -19,6 +19,7 @@ 
 #include "qemu/jhash.h"
 #include "qemu/timer.h"
 #include "slirp/tcp.h"
+#include "qemu/event_notifier.h"
 
 #define HASHTABLE_MAX_SIZE 16384
 
@@ -89,4 +90,22 @@  void connection_hashtable_reset(GHashTable *connection_track_table);
 Packet *packet_new(const void *data, int size);
 void packet_destroy(void *opaque, void *user_data);
 
+typedef void FilterNotifierCallback(void *opaque, int value);
+typedef struct FilterNotifier {
+    GSource source;
+    EventNotifier event;
+    GPollFD pfd;
+    FilterNotifierCallback *cb;
+    void *opaque;
+} FilterNotifier;
+
+FilterNotifier *filter_notifier_new(FilterNotifierCallback *cb,
+                    void *opaque, Error **errp);
+int filter_notifier_set(FilterNotifier *notify, uint64_t value);
+int filter_notifier_get(FilterNotifier *notify, uint64_t *value);
+
+enum {
+    COLO_CHECKPOINT = 2,
+    COLO_FAILOVER,
+};
 #endif /* QEMU_COLO_PROXY_H */