Re: [Qemu-devel] [RFC PATCH v8 10/21] replay: asynchronous events infrastructure

[Paolo Bonzini]

On 22/01/2015 09:52, Pavel Dovgalyuk wrote:
> This patch adds module for saving and replaying asynchronous events.
> These events include network packets, keyboard and mouse input,
> USB packets, thread pool and bottom halves callbacks.
> All events are stored in the queue to be processed at synchronization points
> such as beginning of TB execution, or checkpoint in the iothread.
> 
> Signed-off-by: Pavel Dovgalyuk <address@hidden>
> ---
>  replay/Makefile.objs     |    1 
>  replay/replay-events.c   |  228 
> ++++++++++++++++++++++++++++++++++++++++++++++
>  replay/replay-internal.h |   33 ++++++-
>  replay/replay.h          |    4 +
>  4 files changed, 265 insertions(+), 1 deletions(-)
>  create mode 100755 replay/replay-events.c
> 
> diff --git a/replay/Makefile.objs b/replay/Makefile.objs
> index 1148f45..56da09c 100755
> --- a/replay/Makefile.objs
> +++ b/replay/Makefile.objs
> @@ -1,2 +1,3 @@
>  obj-y += replay.o
>  obj-y += replay-internal.o
> +obj-y += replay-events.o
> diff --git a/replay/replay-events.c b/replay/replay-events.c
> new file mode 100755
> index 0000000..dfd5efd
> --- /dev/null
> +++ b/replay/replay-events.c
> @@ -0,0 +1,228 @@
> +/*
> + * replay-events.c
> + *
> + * Copyright (c) 2010-2015 Institute for System Programming
> + *                         of the Russian Academy of Sciences.
> + *
> + * This work is licensed under the terms of the GNU GPL, version 2 or later.
> + * See the COPYING file in the top-level directory.
> + *
> + */
> +
> +#include "qemu-common.h"
> +#include "replay.h"
> +#include "replay-internal.h"
> +
> +typedef struct Event {
> +    ReplayAsyncEventKind event_kind;
> +    void *opaque;
> +    void *opaque2;
> +    uint64_t id;
> +
> +    QTAILQ_ENTRY(Event) events;
> +} Event;
> +
> +static QTAILQ_HEAD(, Event) events_list = 
> QTAILQ_HEAD_INITIALIZER(events_list);
> +/* Mutex to protect events_list modifications */
> +static QemuMutex lock;

Can you just reuse the replay_lock?  Otherwise you have to document the
lock hierarchy.  But if a single coarser lock is enough, that would be nice.

> +static unsigned int read_event_kind = -1;
> +static uint64_t read_id = -1;
> +static int read_opt = -1;

Please document what "opt" means.

> +
> +static bool replay_events_enabled = false;
> +
> +/* Functions */
> +
> +static void replay_run_event(Event *event)
> +{
> +    switch (event->event_kind) {
> +    default:
> +        fprintf(stderr, "Replay: invalid async event ID (%d) in the queue\n",
> +                event->event_kind);
> +        exit(1);

Please treat this the same as ferror().  fprintf(stderr) should be
replaced by error_report(), which btw takes a string that is not
terminated by \n.

> +        break;
> +    }
> +}
> +
> +void replay_enable_events(void)
> +{
> +    replay_events_enabled = true;
> +}
> +
> +bool replay_has_events(void)
> +{
> +    return !QTAILQ_EMPTY(&events_list);
> +}
> +
> +void replay_flush_events(void)
> +{
> +    qemu_mutex_lock(&lock);
> +    while (!QTAILQ_EMPTY(&events_list)) {
> +        Event *event = QTAILQ_FIRST(&events_list);
> +        qemu_mutex_unlock(&lock);
> +        replay_run_event(event);
> +        qemu_mutex_lock(&lock);
> +        QTAILQ_REMOVE(&events_list, event, events);
> +        g_free(event);
> +    }
> +    qemu_mutex_unlock(&lock);
> +}
> +
> +void replay_disable_events(void)
> +{
> +    replay_events_enabled = false;
> +    /* Flush events queue before waiting of completion */
> +    replay_flush_events();
> +}
> +
> +void replay_clear_events(void)
> +{
> +    qemu_mutex_lock(&lock);
> +    while (!QTAILQ_EMPTY(&events_list)) {
> +        Event *event = QTAILQ_FIRST(&events_list);
> +        QTAILQ_REMOVE(&events_list, event, events);
> +
> +        g_free(event);
> +    }
> +    qemu_mutex_unlock(&lock);
> +}
> +
> +static void replay_add_event_internal(ReplayAsyncEventKind event_kind,
> +                                      void *opaque,
> +                                      void *opaque2, uint64_t id)
> +{
> +    if (event_kind >= REPLAY_ASYNC_COUNT) {
> +        fprintf(stderr, "Replay: invalid async event ID (%d)\n", event_kind);
> +        exit(1);
> +    }
> +    if (!replay_file || replay_mode == REPLAY_MODE_NONE
> +        || !replay_events_enabled) {
> +        Event e;
> +        e.event_kind = event_kind;
> +        e.opaque = opaque;
> +        e.opaque2 = opaque2;
> +        e.id = id;
> +        replay_run_event(&e);
> +        return;
> +    }
> +
> +    Event *event = g_malloc0(sizeof(Event));
> +    event->event_kind = event_kind;
> +    event->opaque = opaque;
> +    event->opaque2 = opaque2;
> +    event->id = id;
> +
> +    qemu_mutex_lock(&lock);
> +    QTAILQ_INSERT_TAIL(&events_list, event, events);
> +    qemu_mutex_unlock(&lock);
> +}
> +
> +void replay_add_event(ReplayAsyncEventKind event_kind, void *opaque)
> +{
> +    replay_add_event_internal(event_kind, opaque, NULL, 0);
> +}
> +
> +/* Called with replay mutex locked */
> +void replay_save_events(int opt)
> +{
> +    qemu_mutex_lock(&lock);
> +    while (!QTAILQ_EMPTY(&events_list)) {
> +        Event *event = QTAILQ_FIRST(&events_list);
> +        if (replay_mode != REPLAY_MODE_PLAY) {
> +            /* put the event into the file */
> +            replay_put_event(EVENT_ASYNC);
> +            replay_put_byte(opt);
> +            replay_put_byte(event->event_kind);
> +
> +            /* save event-specific data */
> +            switch (event->event_kind) {
> +            }

Please extract body of "if" to a separate function replay_save_event.

> +        }
> +
> +        qemu_mutex_unlock(&lock);
> +        replay_mutex_unlock();
> +        replay_run_event(event);
> +        replay_mutex_lock();
> +        qemu_mutex_lock(&lock);
> +        QTAILQ_REMOVE(&events_list, event, events);
> +        g_free(event);
> +    }
> +    qemu_mutex_unlock(&lock);
> +}
> +
> +/* Called with replay mutex locked */
> +void replay_read_events(int opt)
> +{
> +    replay_fetch_data_kind();
> +    while (replay_data_kind == EVENT_ASYNC) {
> +        if (read_event_kind == -1) {
> +            read_opt = replay_get_byte();
> +            read_event_kind = replay_get_byte();
> +            read_id = -1;
> +            replay_check_error();
> +        }
> +
> +        if (opt != read_opt) {
> +            break;
> +        }
> +        /* Execute some events without searching them in the queue */
> +        switch (read_event_kind) {
> +        default:
> +            fprintf(stderr, "Unknown ID %d of replay event\n", 
> read_event_kind);

Again, please treat this the same as ferror().

> +            exit(1);
> +            break;
> +        }
> +
> +        qemu_mutex_lock(&lock);
> +
> +        Event *event = NULL;
> +        Event *curr = NULL;
> +        QTAILQ_FOREACH(curr, &events_list, events) {
> +            if (curr->event_kind == read_event_kind
> +                && (read_id == -1 || read_id == curr->id)) {
> +                event = curr;
> +                break;
> +            }
> +        }
> +
> +        if (event) {
> +            /* read event-specific reading data */
> +
> +            QTAILQ_REMOVE(&events_list, event, events);
> +
> +            qemu_mutex_unlock(&lock);
> +
> +            /* reset unread data and other parameters to allow
> +               reading other data from the log while
> +               running the event */
> +            replay_has_unread_data = 0;
> +            read_event_kind = -1;
> +            read_id = -1;
> +            read_opt = -1;

Everything up to here can be extracted to a separate function that
returns Event *.  Then this function is:

   while (replay_data_kind == EVENT_ASYNC) {
       event = replay_read_event(opt);
       if (!event) {
           break;
       }

       replay_mutex_unlock();
       replay_run_event(event);
       replay_mutex_lock();
       g_free(event);
       replay_fetch_data_kind();
  }

> +            replay_mutex_unlock();
> +            replay_run_event(event);
> +            replay_mutex_lock();
> +            g_free(event);
> +
> +            replay_fetch_data_kind();
> +        } else {
> +            qemu_mutex_unlock(&lock);
> +            /* No such event found in the queue */
> +            break;
> +        }
> +    }
> +}
> +
> +void replay_init_events(void)
> +{
> +    read_event_kind = -1;
> +    qemu_mutex_init(&lock);
> +}
> +
> +void replay_finish_events(void)
> +{
> +    replay_events_enabled = false;
> +    replay_clear_events();
> +    qemu_mutex_destroy(&lock);
> +}
> diff --git a/replay/replay-internal.h b/replay/replay-internal.h
> index 64cc3b2..1666d6e 100755
> --- a/replay/replay-internal.h
> +++ b/replay/replay-internal.h
> @@ -20,9 +20,19 @@ enum ReplayEvents {
>      /* for software interrupt */
>      EVENT_INTERRUPT,
>      /* for emulated exceptions */
> -    EVENT_EXCEPTION
> +    EVENT_EXCEPTION,

Leave a trailing comma, and the patches become nicer. :)

> +    /* for async events */
> +    EVENT_ASYNC

Perhaps add EVENT_COUNT and add a sanity check on load?

Paolo

>  };
>  
> +/* Asynchronous events IDs */
> +
> +enum ReplayAsyncEventKind {
> +    REPLAY_ASYNC_COUNT
> +};
> +
> +typedef enum ReplayAsyncEventKind ReplayAsyncEventKind;
> +
>  typedef struct ReplayState {
>      /*! Current step - number of processed instructions and timer events. */
>      uint64_t current_step;
> @@ -78,4 +88,25 @@ bool skip_async_events(int stop_event);
>      reports an error and stops the execution. */
>  void skip_async_events_until(unsigned int kind);
>  
> +/* Asynchronous events queue */
> +
> +/*! Initializes events' processing internals */
> +void replay_init_events(void);
> +/*! Clears internal data structures for events handling */
> +void replay_finish_events(void);
> +/*! Enables storing events in the queue */
> +void replay_enable_events(void);
> +/*! Flushes events queue */
> +void replay_flush_events(void);
> +/*! Clears events list before loading new VM state */
> +void replay_clear_events(void);
> +/*! Returns true if there are any unsaved events in the queue */
> +bool replay_has_events(void);
> +/*! Saves events from queue into the file */
> +void replay_save_events(int opt);
> +/*! Read events from the file into the input queue */
> +void replay_read_events(int opt);
> +/*! Adds specified async event to the queue */
> +void replay_add_event(ReplayAsyncEventKind event_id, void *opaque);
> +
>  #endif
> diff --git a/replay/replay.h b/replay/replay.h
> index eb3b0ff..31ca3b9 100755
> --- a/replay/replay.h
> +++ b/replay/replay.h
> @@ -43,5 +43,9 @@ bool replay_interrupt(void);
>      Returns true, when interrupt request is pending */
>  bool replay_has_interrupt(void);
>  
> +/* Asynchronous events queue */
> +
> +/*! Disables storing events in the queue */
> +void replay_disable_events(void);
>  
>  #endif
>