[Qemu-devel] [RFC 0/3] monitor: throttle VSERPORT_CHANGED by "id"

[Qemu-devel] [RFC 0/3] monitor: throttle VSERPORT_CHANGED by "id"

[marcandre.lureau]

From: Marc-André Lureau <marcandre.lureau@redhat.com>

QAPI_EVENT_VSERPORT_CHANGE reports changes of a virtio serial port
state. However, the events may be for different ports, but the throttle
mechanism may replace the event for a different port, since it only
checks the event type.

The following series implements throttling of events based on the "id"
field. Hopefully this hash table approach can be later extended if
other fields or combination of fields have to be used.

Marc-André Lureau (3):
  monitor: split MonitorQAPIEventState
  monitor: throttle QAPI_EVENT_VSERPORT_CHANGE by "id"
  monitor: remove old entries from event hash table

 monitor.c    | 245 +++++++++++++++++++++++++++++++++++++++++++++--------------
 trace-events |   2 +-
 2 files changed, 188 insertions(+), 59 deletions(-)

-- 
2.4.3

[Qemu-devel] [RFC 1/3] monitor: split MonitorQAPIEventState

[marcandre.lureau]

From: Marc-André Lureau <marcandre.lureau@redhat.com>

Create a seperate pending event structure MonitorQAPIEventPending.
Use a MonitorQAPIEventDelay callback to handle the delaying. This
allows other implementations of throttling.

Signed-off-by: Marc-André Lureau <marcandre.lureau@redhat.com>
---
 monitor.c    | 124 +++++++++++++++++++++++++++++++++++++----------------------
 trace-events |   2 +-
 2 files changed, 79 insertions(+), 47 deletions(-)

diff --git a/monitor.c b/monitor.c
index aeea2b5..9c51ffa 100644
--- a/monitor.c
+++ b/monitor.c
@@ -170,18 +170,27 @@ typedef struct {
     bool in_command_mode;       /* are we in command mode? */
 } MonitorQMP;
 
+typedef struct MonitorQAPIEventPending {
+    QAPIEvent event;    /* Event being tracked */
+    int64_t last;       /* QEMU_CLOCK_REALTIME value at last emission */
+    QEMUTimer *timer;   /* Timer for handling delayed events */
+    QObject *data;      /* Event pending delayed dispatch */
+} MonitorQAPIEventPending;
+
+typedef struct MonitorQAPIEventState MonitorQAPIEventState;
+
+typedef bool (*MonitorQAPIEventDelay) (MonitorQAPIEventState *evstate,
+                                       QDict *data);
 /*
  * To prevent flooding clients, events can be throttled. The
  * throttling is calculated globally, rather than per-Monitor
  * instance.
  */
-typedef struct MonitorQAPIEventState {
-    QAPIEvent event;    /* Event being tracked */
+struct MonitorQAPIEventState {
     int64_t rate;       /* Minimum time (in ns) between two events */
-    int64_t last;       /* QEMU_CLOCK_REALTIME value at last emission */
-    QEMUTimer *timer;   /* Timer for handling delayed events */
-    QObject *data;      /* Event pending delayed dispatch */
-} MonitorQAPIEventState;
+    MonitorQAPIEventDelay delay;
+    gpointer data;
+};
 
 struct Monitor {
     CharDriverState *chr;
@@ -452,6 +461,39 @@ static void monitor_qapi_event_emit(QAPIEvent event, QObject *data)
     }
 }
 
+static bool
+monitor_qapi_event_delay(MonitorQAPIEventState *evstate, QDict *data)
+{
+    int64_t now = qemu_clock_get_ns(QEMU_CLOCK_REALTIME);
+    MonitorQAPIEventPending *p = evstate->data;
+    int64_t delta = now - p->last;
+
+    /* Rate limit of 0 indicates no throttling */
+    if (!evstate->rate) {
+        p->last = now;
+        return FALSE;
+    }
+
+    if (p->data || delta < evstate->rate) {
+        /* If there's an existing event pending, replace
+         * it with the new event, otherwise schedule a
+         * timer for delayed emission
+         */
+        if (p->data) {
+            qobject_decref(p->data);
+        } else {
+            int64_t then = p->last + evstate->rate;
+            timer_mod_ns(p->timer, then);
+        }
+        p->data = QOBJECT(data);
+        qobject_incref(p->data);
+        return TRUE;
+    }
+
+    p->last = now;
+    return FALSE;
+}
+
 /*
  * Queue a new event for emission to Monitor instances,
  * applying any rate limiting if required.
@@ -467,35 +509,15 @@ monitor_qapi_event_queue(QAPIEvent event, QDict *data, Error **errp)
     trace_monitor_protocol_event_queue(event,
                                        data,
                                        evstate->rate,
-                                       evstate->last,
                                        now);
 
-    /* Rate limit of 0 indicates no throttling */
     qemu_mutex_lock(&monitor_lock);
-    if (!evstate->rate) {
+
+    if (!evstate->delay ||
+        !evstate->delay(evstate, data)) {
         monitor_qapi_event_emit(event, QOBJECT(data));
-        evstate->last = now;
-    } else {
-        int64_t delta = now - evstate->last;
-        if (evstate->data ||
-            delta < evstate->rate) {
-            /* If there's an existing event pending, replace
-             * it with the new event, otherwise schedule a
-             * timer for delayed emission
-             */
-            if (evstate->data) {
-                qobject_decref(evstate->data);
-            } else {
-                int64_t then = evstate->last + evstate->rate;
-                timer_mod_ns(evstate->timer, then);
-            }
-            evstate->data = QOBJECT(data);
-            qobject_incref(evstate->data);
-        } else {
-            monitor_qapi_event_emit(event, QOBJECT(data));
-            evstate->last = now;
-        }
     }
+
     qemu_mutex_unlock(&monitor_lock);
 }
 
@@ -505,23 +527,37 @@ monitor_qapi_event_queue(QAPIEvent event, QDict *data, Error **errp)
  */
 static void monitor_qapi_event_handler(void *opaque)
 {
-    MonitorQAPIEventState *evstate = opaque;
+    MonitorQAPIEventPending *p = opaque;
     int64_t now = qemu_clock_get_ns(QEMU_CLOCK_REALTIME);
 
-    trace_monitor_protocol_event_handler(evstate->event,
-                                         evstate->data,
-                                         evstate->last,
+    trace_monitor_protocol_event_handler(p->event,
+                                         p->data,
+                                         p->last,
                                          now);
     qemu_mutex_lock(&monitor_lock);
-    if (evstate->data) {
-        monitor_qapi_event_emit(evstate->event, evstate->data);
-        qobject_decref(evstate->data);
-        evstate->data = NULL;
+    if (p->data) {
+        monitor_qapi_event_emit(p->event, p->data);
+        qobject_decref(p->data);
+        p->data = NULL;
     }
-    evstate->last = now;
+    p->last = now;
     qemu_mutex_unlock(&monitor_lock);
 }
 
+static MonitorQAPIEventPending *
+monitor_qapi_event_pending_new(QAPIEvent event)
+{
+    MonitorQAPIEventPending *p;
+
+    p = g_new0(MonitorQAPIEventPending, 1);
+    p->event = event;
+    p->timer = timer_new(QEMU_CLOCK_REALTIME,
+                         SCALE_MS,
+                         monitor_qapi_event_handler,
+                         p);
+    return p;
+}
+
 /*
  * @event: the event ID to be limited
  * @rate: the rate limit in milliseconds
@@ -539,15 +575,11 @@ monitor_qapi_event_throttle(QAPIEvent event, int64_t rate)
     evstate = &(monitor_qapi_event_state[event]);
 
     trace_monitor_protocol_event_throttle(event, rate);
-    evstate->event = event;
     assert(rate * SCALE_MS <= INT64_MAX);
     evstate->rate = rate * SCALE_MS;
-    evstate->last = 0;
-    evstate->data = NULL;
-    evstate->timer = timer_new(QEMU_CLOCK_REALTIME,
-                               SCALE_MS,
-                               monitor_qapi_event_handler,
-                               evstate);
+
+    evstate->delay = monitor_qapi_event_delay;
+    evstate->data = monitor_qapi_event_pending_new(event);
 }
 
 static void monitor_qapi_event_init(void)
diff --git a/trace-events b/trace-events
index 94bf3bb..e39ff33 100644
--- a/trace-events
+++ b/trace-events
@@ -1027,7 +1027,7 @@ handle_qmp_command(void *mon, const char *cmd_name) "mon %p cmd_name \"%s\""
 monitor_protocol_emitter(void *mon) "mon %p"
 monitor_protocol_event_handler(uint32_t event, void *data, uint64_t last, uint64_t now) "event=%d data=%p last=%" PRId64 " now=%" PRId64
 monitor_protocol_event_emit(uint32_t event, void *data) "event=%d data=%p"
-monitor_protocol_event_queue(uint32_t event, void *data, uint64_t rate, uint64_t last, uint64_t now) "event=%d data=%p rate=%" PRId64 " last=%" PRId64 " now=%" PRId64
+monitor_protocol_event_queue(uint32_t event, void *data, uint64_t rate, uint64_t now) "event=%d data=%p rate=%" PRId64 " now=%" PRId64
 monitor_protocol_event_throttle(uint32_t event, uint64_t rate) "event=%d rate=%" PRId64
 
 # hw/net/opencores_eth.c
-- 
2.4.3

[Qemu-devel] [RFC 2/3] monitor: throttle QAPI_EVENT_VSERPORT_CHANGE by "id"

[marcandre.lureau]

From: Marc-André Lureau <marcandre.lureau@redhat.com>

Use a hash table to lookup the pending event corresponding to the "id"
field. The hash table may grow without limit here, the following patch
will add some cleaning.

Signed-off-by: Marc-André Lureau <marcandre.lureau@redhat.com>
---
 monitor.c | 104 ++++++++++++++++++++++++++++++++++++++++++++++----------------
 1 file changed, 78 insertions(+), 26 deletions(-)

diff --git a/monitor.c b/monitor.c
index 9c51ffa..7334ffb 100644
--- a/monitor.c
+++ b/monitor.c
@@ -462,10 +462,10 @@ static void monitor_qapi_event_emit(QAPIEvent event, QObject *data)
 }
 
 static bool
-monitor_qapi_event_delay(MonitorQAPIEventState *evstate, QDict *data)
+monitor_qapi_event_pending_update(MonitorQAPIEventState *evstate,
+                                  MonitorQAPIEventPending *p, QDict *data)
 {
     int64_t now = qemu_clock_get_ns(QEMU_CLOCK_REALTIME);
-    MonitorQAPIEventPending *p = evstate->data;
     int64_t delta = now - p->last;
 
     /* Rate limit of 0 indicates no throttling */
@@ -494,31 +494,13 @@ monitor_qapi_event_delay(MonitorQAPIEventState *evstate, QDict *data)
     return FALSE;
 }
 
-/*
- * Queue a new event for emission to Monitor instances,
- * applying any rate limiting if required.
- */
-static void
-monitor_qapi_event_queue(QAPIEvent event, QDict *data, Error **errp)
-{
-    MonitorQAPIEventState *evstate;
-    assert(event < QAPI_EVENT_MAX);
-    int64_t now = qemu_clock_get_ns(QEMU_CLOCK_REALTIME);
-
-    evstate = &(monitor_qapi_event_state[event]);
-    trace_monitor_protocol_event_queue(event,
-                                       data,
-                                       evstate->rate,
-                                       now);
 
-    qemu_mutex_lock(&monitor_lock);
-
-    if (!evstate->delay ||
-        !evstate->delay(evstate, data)) {
-        monitor_qapi_event_emit(event, QOBJECT(data));
-    }
+static bool
+monitor_qapi_event_delay(MonitorQAPIEventState *evstate, QDict *data)
+{
+    MonitorQAPIEventPending *p = evstate->data;
 
-    qemu_mutex_unlock(&monitor_lock);
+    return monitor_qapi_event_pending_update(evstate, p, data);
 }
 
 /*
@@ -558,6 +540,58 @@ monitor_qapi_event_pending_new(QAPIEvent event)
     return p;
 }
 
+static bool
+monitor_qapi_event_id_delay(MonitorQAPIEventState *evstate, QDict *data)
+{
+    GHashTable *ht = evstate->data;
+    QDict *s = qdict_get_qdict(data, "data");
+    const char *id = qdict_get_str(s, "id");
+    MonitorQAPIEventPending *p = g_hash_table_lookup(ht, id);
+    QAPIEvent event = evstate - monitor_qapi_event_state;
+
+    if (!p) {
+        p = monitor_qapi_event_pending_new(event);
+        g_hash_table_insert(ht, g_strdup(id), p);
+    }
+
+    return monitor_qapi_event_pending_update(evstate, p, data);
+}
+
+/*
+ * Queue a new event for emission to Monitor instances,
+ * applying any rate limiting if required.
+ */
+static void
+monitor_qapi_event_queue(QAPIEvent event, QDict *data, Error **errp)
+{
+    MonitorQAPIEventState *evstate;
+    assert(event < QAPI_EVENT_MAX);
+    int64_t now = qemu_clock_get_ns(QEMU_CLOCK_REALTIME);
+
+    evstate = &(monitor_qapi_event_state[event]);
+    trace_monitor_protocol_event_queue(event,
+                                       data,
+                                       evstate->rate,
+                                       now);
+
+    qemu_mutex_lock(&monitor_lock);
+
+    if (!evstate->delay ||
+        !evstate->delay(evstate, data)) {
+        monitor_qapi_event_emit(event, QOBJECT(data));
+    }
+
+    qemu_mutex_unlock(&monitor_lock);
+}
+
+static void
+monitor_qapi_event_pending_free(MonitorQAPIEventPending *p)
+{
+    timer_del(p->timer);
+    timer_free(p->timer);
+    g_free(p);
+}
+
 /*
  * @event: the event ID to be limited
  * @rate: the rate limit in milliseconds
@@ -582,6 +616,24 @@ monitor_qapi_event_throttle(QAPIEvent event, int64_t rate)
     evstate->data = monitor_qapi_event_pending_new(event);
 }
 
+static void
+monitor_qapi_event_id_throttle(QAPIEvent event, int64_t rate)
+{
+    MonitorQAPIEventState *evstate;
+    assert(event < QAPI_EVENT_MAX);
+
+    evstate = &(monitor_qapi_event_state[event]);
+
+    trace_monitor_protocol_event_throttle(event, rate);
+    assert(rate * SCALE_MS <= INT64_MAX);
+    evstate->rate = rate * SCALE_MS;
+
+    evstate->delay = monitor_qapi_event_id_delay;
+    evstate->data =
+        g_hash_table_new_full(g_str_hash, g_str_equal, NULL,
+            (GDestroyNotify)monitor_qapi_event_pending_free);
+}
+
 static void monitor_qapi_event_init(void)
 {
     /* Limit guest-triggerable events to 1 per second */
@@ -590,7 +642,7 @@ static void monitor_qapi_event_init(void)
     monitor_qapi_event_throttle(QAPI_EVENT_BALLOON_CHANGE, 1000);
     monitor_qapi_event_throttle(QAPI_EVENT_QUORUM_REPORT_BAD, 1000);
     monitor_qapi_event_throttle(QAPI_EVENT_QUORUM_FAILURE, 1000);
-    monitor_qapi_event_throttle(QAPI_EVENT_VSERPORT_CHANGE, 1000);
+    monitor_qapi_event_id_throttle(QAPI_EVENT_VSERPORT_CHANGE, 1000);
 
     qmp_event_set_func_emit(monitor_qapi_event_queue);
 }
-- 
2.4.3

[Qemu-devel] [RFC 3/3] monitor: remove old entries from event hash table

[marcandre.lureau]

From: Marc-André Lureau <marcandre.lureau@redhat.com>

Do not let the hash table grow without limit, schedule a cleanup for
outdated event.

Signed-off-by: Marc-André Lureau <marcandre.lureau@redhat.com>
---
 monitor.c | 51 ++++++++++++++++++++++++++++++++++++++++++++++++---
 1 file changed, 48 insertions(+), 3 deletions(-)

diff --git a/monitor.c b/monitor.c
index 7334ffb..6980806 100644
--- a/monitor.c
+++ b/monitor.c
@@ -170,6 +170,8 @@ typedef struct {
     bool in_command_mode;       /* are we in command mode? */
 } MonitorQMP;
 
+typedef struct MonitorQAPIEventState MonitorQAPIEventState;
+
 typedef struct MonitorQAPIEventPending {
     QAPIEvent event;    /* Event being tracked */
     int64_t last;       /* QEMU_CLOCK_REALTIME value at last emission */
@@ -177,8 +179,6 @@ typedef struct MonitorQAPIEventPending {
     QObject *data;      /* Event pending delayed dispatch */
 } MonitorQAPIEventPending;
 
-typedef struct MonitorQAPIEventState MonitorQAPIEventState;
-
 typedef bool (*MonitorQAPIEventDelay) (MonitorQAPIEventState *evstate,
                                        QDict *data);
 /*
@@ -540,6 +540,45 @@ monitor_qapi_event_pending_new(QAPIEvent event)
     return p;
 }
 
+static void monitor_qapi_event_id_remove(void *opaque)
+{
+    MonitorQAPIEventPending *p = opaque;
+    MonitorQAPIEventState *s = &monitor_qapi_event_state[p->event];
+    GHashTable *ht = s->data;
+    GHashTableIter iter;
+    gpointer value;
+
+    g_hash_table_iter_init(&iter, ht);
+    while (g_hash_table_iter_next(&iter, NULL, &value)) {
+        if (value == p) {
+            g_hash_table_iter_remove(&iter);
+            return;
+        }
+    }
+}
+
+/*
+ * do not let the hash table grow, if no later pending event
+ * scheduled, remove the old entry after rate timeout.
+ */
+static void monitor_qapi_event_id_schedule_remove(MonitorQAPIEventPending *p)
+{
+    MonitorQAPIEventState *s = &monitor_qapi_event_state[p->event];
+    int64_t now = qemu_clock_get_ns(QEMU_CLOCK_REALTIME);
+    int64_t then = now + s->rate;
+
+    p->timer->cb = monitor_qapi_event_id_remove;
+    timer_mod_ns(p->timer, then);
+}
+
+static void monitor_qapi_event_id_handler(void *opaque)
+{
+    MonitorQAPIEventPending *p = opaque;
+
+    monitor_qapi_event_handler(p);
+    monitor_qapi_event_id_schedule_remove(p);
+}
+
 static bool
 monitor_qapi_event_id_delay(MonitorQAPIEventState *evstate, QDict *data)
 {
@@ -554,7 +593,13 @@ monitor_qapi_event_id_delay(MonitorQAPIEventState *evstate, QDict *data)
         g_hash_table_insert(ht, g_strdup(id), p);
     }
 
-    return monitor_qapi_event_pending_update(evstate, p, data);
+    if (monitor_qapi_event_pending_update(evstate, p, data)) {
+        p->timer->cb = monitor_qapi_event_id_handler;
+        return TRUE;
+    } else {
+        monitor_qapi_event_id_schedule_remove(p);
+        return FALSE;
+    }
 }
 
 /*
-- 
2.4.3

Re: [Qemu-devel] [RFC 1/3] monitor: split MonitorQAPIEventState

[Laszlo Ersek]

On 08/12/15 21:46, marcandre.lureau@redhat.com wrote:
> From: Marc-André Lureau <marcandre.lureau@redhat.com>
> 
> Create a seperate pending event structure MonitorQAPIEventPending.
> Use a MonitorQAPIEventDelay callback to handle the delaying. This
> allows other implementations of throttling.
> 
> Signed-off-by: Marc-André Lureau <marcandre.lureau@redhat.com>
> ---
>  monitor.c    | 124 +++++++++++++++++++++++++++++++++++++----------------------
>  trace-events |   2 +-
>  2 files changed, 79 insertions(+), 47 deletions(-)

Assume there has been a long period of silence (no attempts to emit an
event). Now some client code makes a call to emit the event.

Will that event be emitted immediately, or will it be delayed to see if
more are coming? I'd like to understand this aspect first.

I think the first instance of the event, after the grace period, should
be emitted immediately, and further instances that quickly follow should
be suppressed.

It would also be possible to delay (and update) a "current instance"
in-place until the calls quiesce, and emit the (most recent) update
then. (I'm getting the impression that the patch does this, but I could
be wrong.)

I think emitting the first immediately (and suppressing followers) is
better; that's eg. what the kernel log does.

Nagle's algorithm (TCP_CORK <-> TCP_NODELAY) is a (strained) example for
the other behavior (ie. coalesce leaders). I think that approach would
not be a natural fit for QEMU's events.

Of course, if your code already implements what I think would be more
applicable, then I shouldn't have spoken! :) Can you clarify it though
please?

Thanks!
Laszlo


> diff --git a/monitor.c b/monitor.c
> index aeea2b5..9c51ffa 100644
> --- a/monitor.c
> +++ b/monitor.c
> @@ -170,18 +170,27 @@ typedef struct {
>      bool in_command_mode;       /* are we in command mode? */
>  } MonitorQMP;
>  
> +typedef struct MonitorQAPIEventPending {
> +    QAPIEvent event;    /* Event being tracked */
> +    int64_t last;       /* QEMU_CLOCK_REALTIME value at last emission */
> +    QEMUTimer *timer;   /* Timer for handling delayed events */
> +    QObject *data;      /* Event pending delayed dispatch */
> +} MonitorQAPIEventPending;
> +
> +typedef struct MonitorQAPIEventState MonitorQAPIEventState;
> +
> +typedef bool (*MonitorQAPIEventDelay) (MonitorQAPIEventState *evstate,
> +                                       QDict *data);
>  /*
>   * To prevent flooding clients, events can be throttled. The
>   * throttling is calculated globally, rather than per-Monitor
>   * instance.
>   */
> -typedef struct MonitorQAPIEventState {
> -    QAPIEvent event;    /* Event being tracked */
> +struct MonitorQAPIEventState {
>      int64_t rate;       /* Minimum time (in ns) between two events */
> -    int64_t last;       /* QEMU_CLOCK_REALTIME value at last emission */
> -    QEMUTimer *timer;   /* Timer for handling delayed events */
> -    QObject *data;      /* Event pending delayed dispatch */
> -} MonitorQAPIEventState;
> +    MonitorQAPIEventDelay delay;
> +    gpointer data;
> +};
>  
>  struct Monitor {
>      CharDriverState *chr;
> @@ -452,6 +461,39 @@ static void monitor_qapi_event_emit(QAPIEvent event, QObject *data)
>      }
>  }
>  
> +static bool
> +monitor_qapi_event_delay(MonitorQAPIEventState *evstate, QDict *data)
> +{
> +    int64_t now = qemu_clock_get_ns(QEMU_CLOCK_REALTIME);
> +    MonitorQAPIEventPending *p = evstate->data;
> +    int64_t delta = now - p->last;
> +
> +    /* Rate limit of 0 indicates no throttling */
> +    if (!evstate->rate) {
> +        p->last = now;
> +        return FALSE;
> +    }
> +
> +    if (p->data || delta < evstate->rate) {
> +        /* If there's an existing event pending, replace
> +         * it with the new event, otherwise schedule a
> +         * timer for delayed emission
> +         */
> +        if (p->data) {
> +            qobject_decref(p->data);
> +        } else {
> +            int64_t then = p->last + evstate->rate;
> +            timer_mod_ns(p->timer, then);
> +        }
> +        p->data = QOBJECT(data);
> +        qobject_incref(p->data);
> +        return TRUE;
> +    }
> +
> +    p->last = now;
> +    return FALSE;
> +}
> +
>  /*
>   * Queue a new event for emission to Monitor instances,
>   * applying any rate limiting if required.
> @@ -467,35 +509,15 @@ monitor_qapi_event_queue(QAPIEvent event, QDict *data, Error **errp)
>      trace_monitor_protocol_event_queue(event,
>                                         data,
>                                         evstate->rate,
> -                                       evstate->last,
>                                         now);
>  
> -    /* Rate limit of 0 indicates no throttling */
>      qemu_mutex_lock(&monitor_lock);
> -    if (!evstate->rate) {
> +
> +    if (!evstate->delay ||
> +        !evstate->delay(evstate, data)) {
>          monitor_qapi_event_emit(event, QOBJECT(data));
> -        evstate->last = now;
> -    } else {
> -        int64_t delta = now - evstate->last;
> -        if (evstate->data ||
> -            delta < evstate->rate) {
> -            /* If there's an existing event pending, replace
> -             * it with the new event, otherwise schedule a
> -             * timer for delayed emission
> -             */
> -            if (evstate->data) {
> -                qobject_decref(evstate->data);
> -            } else {
> -                int64_t then = evstate->last + evstate->rate;
> -                timer_mod_ns(evstate->timer, then);
> -            }
> -            evstate->data = QOBJECT(data);
> -            qobject_incref(evstate->data);
> -        } else {
> -            monitor_qapi_event_emit(event, QOBJECT(data));
> -            evstate->last = now;
> -        }
>      }
> +
>      qemu_mutex_unlock(&monitor_lock);
>  }
>  
> @@ -505,23 +527,37 @@ monitor_qapi_event_queue(QAPIEvent event, QDict *data, Error **errp)
>   */
>  static void monitor_qapi_event_handler(void *opaque)
>  {
> -    MonitorQAPIEventState *evstate = opaque;
> +    MonitorQAPIEventPending *p = opaque;
>      int64_t now = qemu_clock_get_ns(QEMU_CLOCK_REALTIME);
>  
> -    trace_monitor_protocol_event_handler(evstate->event,
> -                                         evstate->data,
> -                                         evstate->last,
> +    trace_monitor_protocol_event_handler(p->event,
> +                                         p->data,
> +                                         p->last,
>                                           now);
>      qemu_mutex_lock(&monitor_lock);
> -    if (evstate->data) {
> -        monitor_qapi_event_emit(evstate->event, evstate->data);
> -        qobject_decref(evstate->data);
> -        evstate->data = NULL;
> +    if (p->data) {
> +        monitor_qapi_event_emit(p->event, p->data);
> +        qobject_decref(p->data);
> +        p->data = NULL;
>      }
> -    evstate->last = now;
> +    p->last = now;
>      qemu_mutex_unlock(&monitor_lock);
>  }
>  
> +static MonitorQAPIEventPending *
> +monitor_qapi_event_pending_new(QAPIEvent event)
> +{
> +    MonitorQAPIEventPending *p;
> +
> +    p = g_new0(MonitorQAPIEventPending, 1);
> +    p->event = event;
> +    p->timer = timer_new(QEMU_CLOCK_REALTIME,
> +                         SCALE_MS,
> +                         monitor_qapi_event_handler,
> +                         p);
> +    return p;
> +}
> +
>  /*
>   * @event: the event ID to be limited
>   * @rate: the rate limit in milliseconds
> @@ -539,15 +575,11 @@ monitor_qapi_event_throttle(QAPIEvent event, int64_t rate)
>      evstate = &(monitor_qapi_event_state[event]);
>  
>      trace_monitor_protocol_event_throttle(event, rate);
> -    evstate->event = event;
>      assert(rate * SCALE_MS <= INT64_MAX);
>      evstate->rate = rate * SCALE_MS;
> -    evstate->last = 0;
> -    evstate->data = NULL;
> -    evstate->timer = timer_new(QEMU_CLOCK_REALTIME,
> -                               SCALE_MS,
> -                               monitor_qapi_event_handler,
> -                               evstate);
> +
> +    evstate->delay = monitor_qapi_event_delay;
> +    evstate->data = monitor_qapi_event_pending_new(event);
>  }
>  
>  static void monitor_qapi_event_init(void)
> diff --git a/trace-events b/trace-events
> index 94bf3bb..e39ff33 100644
> --- a/trace-events
> +++ b/trace-events
> @@ -1027,7 +1027,7 @@ handle_qmp_command(void *mon, const char *cmd_name) "mon %p cmd_name \"%s\""
>  monitor_protocol_emitter(void *mon) "mon %p"
>  monitor_protocol_event_handler(uint32_t event, void *data, uint64_t last, uint64_t now) "event=%d data=%p last=%" PRId64 " now=%" PRId64
>  monitor_protocol_event_emit(uint32_t event, void *data) "event=%d data=%p"
> -monitor_protocol_event_queue(uint32_t event, void *data, uint64_t rate, uint64_t last, uint64_t now) "event=%d data=%p rate=%" PRId64 " last=%" PRId64 " now=%" PRId64
> +monitor_protocol_event_queue(uint32_t event, void *data, uint64_t rate, uint64_t now) "event=%d data=%p rate=%" PRId64 " now=%" PRId64
>  monitor_protocol_event_throttle(uint32_t event, uint64_t rate) "event=%d rate=%" PRId64
>  
>  # hw/net/opencores_eth.c
> 

Re: [Qemu-devel] [RFC 1/3] monitor: split MonitorQAPIEventState

[Marc-André Lureau]

Hi

----- Original Message -----
> On 08/12/15 21:46, marcandre.lureau@redhat.com wrote:
> > From: Marc-André Lureau <marcandre.lureau@redhat.com>
> > 
> > Create a seperate pending event structure MonitorQAPIEventPending.
> > Use a MonitorQAPIEventDelay callback to handle the delaying. This
> > allows other implementations of throttling.
> > 
> > Signed-off-by: Marc-André Lureau <marcandre.lureau@redhat.com>
> > ---
> >  monitor.c    | 124
> >  +++++++++++++++++++++++++++++++++++++----------------------
> >  trace-events |   2 +-
> >  2 files changed, 79 insertions(+), 47 deletions(-)
> 
> Assume there has been a long period of silence (no attempts to emit an
> event). Now some client code makes a call to emit the event.
> 
> Will that event be emitted immediately, or will it be delayed to see if
> more are coming? I'd like to understand this aspect first.
> 
> I think the first instance of the event, after the grace period, should
> be emitted immediately, and further instances that quickly follow should
> be suppressed.

This is what qemu already does. The first event is sent immediately, the
later ones may be delayed (but there will be at least one event every period,
if the event is flooded). This patch 1/3 doesn't change the logic, only
it split things to make them a bit more modular.

So the rest of the patches do not change the qemu delay logic, it adds a way to
delay based on (event + "id") instead of just (event). It does that by
adding an additional "id" hashtable for the event type. My hope is
that this apporach could be reused if other field or combinations of fields
are necessary for other events, but for now, it's hardcoded for "id". 

cheers

Re: [Qemu-devel] [RFC 1/3] monitor: split MonitorQAPIEventState

[Eric Blake]

[-- Attachment #1: Type: text/plain, Size: 1873 bytes --]

On 08/12/2015 02:00 PM, Laszlo Ersek wrote:
> Assume there has been a long period of silence (no attempts to emit an
> event). Now some client code makes a call to emit the event.
> 
> Will that event be emitted immediately, or will it be delayed to see if
> more are coming? I'd like to understand this aspect first.
> 
> I think the first instance of the event, after the grace period, should
> be emitted immediately, and further instances that quickly follow should
> be suppressed.

That has always been the goal of event throttling: when a new event
arrives and the timer is not running, emit it right away and start the
timer. If another event arrives while the timer has not expired, save
it.  If multiple events arrive during the timer, the last one received
overwrites any others.  Then, when the timer expires, either there is a
saved event (one or more events arrived during the throttling window),
so we send it and restart another throttling window, or there was no
event pending so the line is quiet and the next event can send right
away.  Thus, management will get notification as soon as possible that
events are starting to happen after a period of quiet, but will then not
be spammed any faster than once per throttle period (once per second)
with additional events.  When additional events are sent, they are
always the most accurate state of the system at the time the event was
queued, but intermediate events may have been lost.  Furthermore, unless
the events have not been happening, the throttling of the event means
the management might not see the event until nearly a second late,
although the timestamp associated with the event should still be
accurate to the time it was queued up, not finally sent.

-- 
Eric Blake   eblake redhat com    +1-919-301-3266
Libvirt virtualization library http://libvirt.org


[-- Attachment #2: OpenPGP digital signature --]
[-- Type: application/pgp-signature, Size: 604 bytes --]

Re: [Qemu-devel] [RFC 1/3] monitor: split MonitorQAPIEventState

[Eric Blake]

[-- Attachment #1: Type: text/plain, Size: 2629 bytes --]

On 08/12/2015 01:46 PM, marcandre.lureau@redhat.com wrote:
> From: Marc-André Lureau <marcandre.lureau@redhat.com>
> 
> Create a seperate pending event structure MonitorQAPIEventPending.

s/seperate/separate/

> Use a MonitorQAPIEventDelay callback to handle the delaying. This
> allows other implementations of throttling.
> 
> Signed-off-by: Marc-André Lureau <marcandre.lureau@redhat.com>
> ---
>  monitor.c    | 124 +++++++++++++++++++++++++++++++++++++----------------------
>  trace-events |   2 +-
>  2 files changed, 79 insertions(+), 47 deletions(-)
> 

>  
> +typedef struct MonitorQAPIEventPending {
> +    QAPIEvent event;    /* Event being tracked */
> +    int64_t last;       /* QEMU_CLOCK_REALTIME value at last emission */
> +    QEMUTimer *timer;   /* Timer for handling delayed events */
> +    QObject *data;      /* Event pending delayed dispatch */
> +} MonitorQAPIEventPending;
> +
> +typedef struct MonitorQAPIEventState MonitorQAPIEventState;

Some places combine the typedef with the struct definition; I'm not sure
there's any hard and fast rule, though. (HACKING mentions that we want
the typedef, but doesn't give guidelines on how it must be provided).

> +struct MonitorQAPIEventState {
>      int64_t rate;       /* Minimum time (in ns) between two events */
> -    int64_t last;       /* QEMU_CLOCK_REALTIME value at last emission */
> -    QEMUTimer *timer;   /* Timer for handling delayed events */
> -    QObject *data;      /* Event pending delayed dispatch */
> -} MonitorQAPIEventState;
> +    MonitorQAPIEventDelay delay;
> +    gpointer data;

Do we really need 'gpointer', or is 'void *' sufficient?

>  
> +static bool
> +monitor_qapi_event_delay(MonitorQAPIEventState *evstate, QDict *data)
> +{
> +    int64_t now = qemu_clock_get_ns(QEMU_CLOCK_REALTIME);
> +    MonitorQAPIEventPending *p = evstate->data;
> +    int64_t delta = now - p->last;
> +
> +    /* Rate limit of 0 indicates no throttling */
> +    if (!evstate->rate) {
> +        p->last = now;
> +        return FALSE;

s/FALSE/false/ (we want to directly use the C99 'bool' type here, not
the glib macros that expand to who-knows-what-type followed by implicit
conversion back to bool).

> +        p->data = QOBJECT(data);
> +        qobject_incref(p->data);
> +        return TRUE;
> +    }
> +
> +    p->last = now;
> +    return FALSE;

two more ALL_CAPS to convert to the lower bool counterpart.

Otherwise looks like a sane split.

-- 
Eric Blake   eblake redhat com    +1-919-301-3266
Libvirt virtualization library http://libvirt.org


[-- Attachment #2: OpenPGP digital signature --]
[-- Type: application/pgp-signature, Size: 604 bytes --]

Re: [Qemu-devel] [RFC 2/3] monitor: throttle QAPI_EVENT_VSERPORT_CHANGE by "id"

[Eric Blake]

[-- Attachment #1: Type: text/plain, Size: 674 bytes --]

On 08/12/2015 01:46 PM, marcandre.lureau@redhat.com wrote:
> From: Marc-André Lureau <marcandre.lureau@redhat.com>
> 
> Use a hash table to lookup the pending event corresponding to the "id"
> field. The hash table may grow without limit here, the following patch
> will add some cleaning.
> 
> Signed-off-by: Marc-André Lureau <marcandre.lureau@redhat.com>
> ---
>  monitor.c | 104 ++++++++++++++++++++++++++++++++++++++++++++++----------------
>  1 file changed, 78 insertions(+), 26 deletions(-)
> 

Reviewed-by: Eric Blake <eblake@redhat.com>

-- 
Eric Blake   eblake redhat com    +1-919-301-3266
Libvirt virtualization library http://libvirt.org


[-- Attachment #2: OpenPGP digital signature --]
[-- Type: application/pgp-signature, Size: 604 bytes --]

Re: [Qemu-devel] [RFC 3/3] monitor: remove old entries from event hash table

[Eric Blake]

[-- Attachment #1: Type: text/plain, Size: 1155 bytes --]

On 08/12/2015 01:46 PM, marcandre.lureau@redhat.com wrote:
> From: Marc-André Lureau <marcandre.lureau@redhat.com>
> 
> Do not let the hash table grow without limit, schedule a cleanup for
> outdated event.
> 
> Signed-off-by: Marc-André Lureau <marcandre.lureau@redhat.com>
> ---
>  monitor.c | 51 ++++++++++++++++++++++++++++++++++++++++++++++++---
>  1 file changed, 48 insertions(+), 3 deletions(-)
> 

> @@ -554,7 +593,13 @@ monitor_qapi_event_id_delay(MonitorQAPIEventState *evstate, QDict *data)
>          g_hash_table_insert(ht, g_strdup(id), p);
>      }
>  
> -    return monitor_qapi_event_pending_update(evstate, p, data);
> +    if (monitor_qapi_event_pending_update(evstate, p, data)) {
> +        p->timer->cb = monitor_qapi_event_id_handler;
> +        return TRUE;
> +    } else {
> +        monitor_qapi_event_id_schedule_remove(p);
> +        return FALSE;

Again, prefer 'true' and 'false' for a function that returns bool.
Otherwise, looks okay to me, and looking forward to a non-RFC version.

-- 
Eric Blake   eblake redhat com    +1-919-301-3266
Libvirt virtualization library http://libvirt.org


[-- Attachment #2: OpenPGP digital signature --]
[-- Type: application/pgp-signature, Size: 604 bytes --]