[RFC,v8,10/21] replay: asynchronous events infrastructure

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 <pavel.dovgaluk@ispras.ru>
---
 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

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 <pavel.dovgaluk@ispras.ru>
> ---
>  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
>