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

[Markus Armbruster <armbru@redhat.com>]

marcandre.lureau@redhat.com writes:

> From: Marc-André Lureau <marcandre.lureau@redhat.com>
>
> Create a separate 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 fc32f12..0a6a16a 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 {
> +    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;

"MonitorQAPIEventPending" sounds like the name of a predicate "is an
event pending?"  MonitorQAPIPendingEvent?

> +
> +typedef struct MonitorQAPIEventState MonitorQAPIEventState;
> +
> +typedef bool (*MonitorQAPIEventDelay) (MonitorQAPIEventState *evstate,
> +                                       QDict *data);

Style nit: we don't normally have space before an argument list.

>  /*
>   * 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;

Since your commit message is so sparse, I have to actually think to
figure out how this patch works.  To think, I have to write.  And when I
write, I can just as well send it out, so you can correct my thinking.

If you want less verbose reviews from me, try writing more verbose
commit messages :)

Obvious: event, last, timer and data move into MonitorQAPIEventDelay.

Not obvious (yet): how to reach them.  Read on and see, I guess.

> +    void *data;

What does data point to?

Why does it have to be void *?

> +};
>  
>  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;
> +}
> +

Okay, this is the part of monitor_qapi_event_queue() protected by the
lock, less the monitor_qapi_event_emit(), with the move of last, time
and data from evstate to p squashed in, and the computation of delta
moved up some.  Would be easier to review if you did the transformations
in separate patches.

Could use a function comment, because the return value isn't obvious.

Also: too many things are named "data" for my taste.

>  /*
>   * 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);

Should monitor_qapi_event_delay() trace evstate->last to compensate?

>  
> -    /* Rate limit of 0 indicates no throttling */
>      qemu_mutex_lock(&monitor_lock);
> -    if (!evstate->rate) {
> +
> +    if (!evstate->delay ||
> +        !evstate->delay(evstate, data)) {

Sure evstate->delay can be null?

>          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)

All right, the moved members end up in evstate->data, and we now
indirect the interesting part of monitor_qapi_event_queue() through
evstate->delay.

Why this is useful isn't obvious, yet.  The only clue I have is the
commit message's "This allows other implementations of throttling."  I'd
add something like "The next few commits will add one."

> diff --git a/trace-events b/trace-events
> index 8f9614a..b1ea596 100644
> --- a/trace-events
> +++ b/trace-events
> @@ -1028,7 +1028,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