Re: [PATCH v4 6/6] blkdebug: protect rules and suspended_reqs with a lock

[Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>]

04.06.2021 13:07, Emanuele Giuseppe Esposito wrote:
> First, categorize the structure fields to identify what needs
> to be protected and what doesn't.
> 
> We essentially need to protect only .state, and the 3 lists in
> BDRVBlkdebugState.
> 
> Then, add the lock and mark the functions accordingly.
> 
> Co-developed-by: Paolo Bonzini <pbonzini@redhat.com>
> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
> ---
>   block/blkdebug.c | 46 +++++++++++++++++++++++++++++++++++-----------
>   1 file changed, 35 insertions(+), 11 deletions(-)
> 
> diff --git a/block/blkdebug.c b/block/blkdebug.c
> index d597753139..ac3799f739 100644
> --- a/block/blkdebug.c
> +++ b/block/blkdebug.c
> @@ -38,24 +38,27 @@
>   #include "qapi/qobject-input-visitor.h"
>   #include "sysemu/qtest.h"
>   
> +/* All APIs are thread-safe */
> +
>   typedef struct BDRVBlkdebugState {
> -    int state;
> +    /* IN: initialized in blkdebug_open() and never changed */
>       uint64_t align;
>       uint64_t max_transfer;
>       uint64_t opt_write_zero;
>       uint64_t max_write_zero;
>       uint64_t opt_discard;
>       uint64_t max_discard;
> -
> +    char *config_file; /* For blkdebug_refresh_filename() */
> +    /* initialized in blkdebug_parse_perms() */
>       uint64_t take_child_perms;
>       uint64_t unshare_child_perms;
>   
> -    /* For blkdebug_refresh_filename() */
> -    char *config_file;
> -
> +    /* State. Protected by lock */
> +    int state;
>       QLIST_HEAD(, BlkdebugRule) rules[BLKDBG__MAX];
>       QSIMPLEQ_HEAD(, BlkdebugRule) active_rules;
>       QLIST_HEAD(, BlkdebugSuspendedReq) suspended_reqs;
> +    QemuMutex lock;
>   } BDRVBlkdebugState;
>   
>   typedef struct BlkdebugAIOCB {
> @@ -64,6 +67,7 @@ typedef struct BlkdebugAIOCB {
>   } BlkdebugAIOCB;
>   
>   typedef struct BlkdebugSuspendedReq {
> +    /* IN: initialized in suspend_request() */
>       Coroutine *co;
>       char *tag;

@next is part of *suspended_reqs list (in a manner), so it should be protected by lock

>       QLIST_ENTRY(BlkdebugSuspendedReq) next;
> @@ -77,6 +81,7 @@ enum {
>   };
>   
>   typedef struct BlkdebugRule {
> +    /* IN: initialized in add_rule() or blkdebug_debug_breakpoint() */
>       BlkdebugEvent event;
>       int action;
>       int state;

as well as @next and @active_next here.

> @@ -244,11 +249,14 @@ static int add_rule(void *opaque, QemuOpts *opts, Error **errp)


>       };
>   
>       /* Add the rule */
> +    qemu_mutex_lock(&s->lock);
>       QLIST_INSERT_HEAD(&s->rules[event], rule, next);
> +    qemu_mutex_unlock(&s->lock);
>   

actually, add_rule is called only from .open(), so doesn't need a lock.. But it doesn't hurt.

>       return 0;
>   }
>   
> +/* Called with lock held or from .bdrv_close */
>   static void remove_rule(BlkdebugRule *rule)
>   {
>       switch (rule->action) {
> @@ -467,6 +475,7 @@ static int blkdebug_open(BlockDriverState *bs, QDict *options, int flags,
>       int ret;
>       uint64_t align;
>   
> +    qemu_mutex_init(&s->lock);
>       opts = qemu_opts_create(&runtime_opts, NULL, 0, &error_abort);
>       if (!qemu_opts_absorb_qdict(opts, options, errp)) {
>           ret = -EINVAL;
> @@ -567,6 +576,7 @@ static int blkdebug_open(BlockDriverState *bs, QDict *options, int flags,
>       ret = 0;
>   out:
>       if (ret < 0) {
> +        qemu_mutex_destroy(&s->lock);
>           g_free(s->config_file);
>       }
>       qemu_opts_del(opts);
> @@ -581,6 +591,7 @@ static int rule_check(BlockDriverState *bs, uint64_t offset, uint64_t bytes,
>       int error;
>       bool immediately;
>   
> +    qemu_mutex_lock(&s->lock);
>       QSIMPLEQ_FOREACH(rule, &s->active_rules, active_next) {
>           uint64_t inject_offset = rule->options.inject.offset;
>   
> @@ -594,6 +605,7 @@ static int rule_check(BlockDriverState *bs, uint64_t offset, uint64_t bytes,
>       }
>   
>       if (!rule || !rule->options.inject.error) {
> +        qemu_mutex_unlock(&s->lock);
>           return 0;
>       }
>   
> @@ -605,6 +617,7 @@ static int rule_check(BlockDriverState *bs, uint64_t offset, uint64_t bytes,
>           remove_rule(rule);
>       }
>   
> +    qemu_mutex_unlock(&s->lock);
>       if (!immediately) {
>           aio_co_schedule(qemu_get_current_aio_context(), qemu_coroutine_self());
>           qemu_coroutine_yield();
> @@ -770,8 +783,10 @@ static void blkdebug_close(BlockDriverState *bs)
>       }
>   
>       g_free(s->config_file);
> +    qemu_mutex_destroy(&s->lock);
>   }
>   
> +/* Called with lock held.  */
>   static void suspend_request(BlockDriverState *bs, BlkdebugRule *rule)
>   {
>       BDRVBlkdebugState *s = bs->opaque;
> @@ -790,6 +805,7 @@ static void suspend_request(BlockDriverState *bs, BlkdebugRule *rule)
>       }
>   }
>   
> +/* Called with lock held.  */
>   static void process_rule(BlockDriverState *bs, struct BlkdebugRule *rule,
>                            int *action_count, int *new_state)
>   {
> @@ -830,17 +846,18 @@ static void blkdebug_debug_event(BlockDriverState *bs, BlkdebugEvent event)
>   
>       assert((int)event >= 0 && event < BLKDBG__MAX);
>   
> -    new_state = s->state;
> -    QLIST_FOREACH_SAFE(rule, &s->rules[event], next, next) {
> -        process_rule(bs, rule, actions_count, &new_state);
> +    WITH_QEMU_LOCK_GUARD(&s->lock) {
> +        new_state = s->state;
> +        QLIST_FOREACH_SAFE(rule, &s->rules[event], next, next) {
> +            process_rule(bs, rule, actions_count, &new_state);
> +        }
> +        s->state = new_state;
>       }
>   
>       while (actions_count[ACTION_SUSPEND] > 0) {
>           qemu_coroutine_yield();
>           actions_count[ACTION_SUSPEND]--;
>       }
> -
> -    s->state = new_state;

Not sure, are all existing users prepared to state update moved to be before actual suspend. But that looks better and as we discussed is safer. So, if all iotests pass, it's OK.

>   }
>   
>   static int blkdebug_debug_breakpoint(BlockDriverState *bs, const char *event,
> @@ -863,11 +880,14 @@ static int blkdebug_debug_breakpoint(BlockDriverState *bs, const char *event,
>           .options.suspend.tag = g_strdup(tag),
>       };
>   
> +    qemu_mutex_lock(&s->lock);
>       QLIST_INSERT_HEAD(&s->rules[blkdebug_event], rule, next);
> +    qemu_mutex_unlock(&s->lock);
>   
>       return 0;
>   }
>   
> +/* Called with lock held. May temporarily release lock. */
>   static int resume_req_by_tag(BDRVBlkdebugState *s, const char *tag, bool all)
>   {
>       BlkdebugSuspendedReq *r;
> @@ -885,7 +905,9 @@ retry:
>               g_free(r->tag);
>               g_free(r);
>   
> +            qemu_mutex_unlock(&s->lock);
>               qemu_coroutine_enter(co);
> +            qemu_mutex_lock(&s->lock);
>   
>               if (all) {
>                   goto retry;
> @@ -899,7 +921,7 @@ retry:
>   static int blkdebug_debug_resume(BlockDriverState *bs, const char *tag)
>   {
>       BDRVBlkdebugState *s = bs->opaque;
> -
> +    QEMU_LOCK_GUARD(&s->lock);
>       return resume_req_by_tag(s, tag, false);
>   }
>   
> @@ -910,6 +932,7 @@ static int blkdebug_debug_remove_breakpoint(BlockDriverState *bs,
>       BlkdebugRule *rule, *next;
>       int i, ret = -ENOENT;
>   
> +    QEMU_LOCK_GUARD(&s->lock);
>       for (i = 0; i < BLKDBG__MAX; i++) {
>           QLIST_FOREACH_SAFE(rule, &s->rules[i], next, next) {
>               if (rule->action == ACTION_SUSPEND &&
> @@ -930,6 +953,7 @@ static bool blkdebug_debug_is_suspended(BlockDriverState *bs, const char *tag)
>       BDRVBlkdebugState *s = bs->opaque;
>       BlkdebugSuspendedReq *r;
>   
> +    QEMU_LOCK_GUARD(&s->lock);
>       QLIST_FOREACH(r, &s->suspended_reqs, next) {
>           if (!strcmp(r->tag, tag)) {
>               return true;
> 


Reviewed-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>


Optional suggestion for additional comments and more use of QEMU_LOCK_GUARD (it looks large because of indentation change):

diff --git a/block/blkdebug.c b/block/blkdebug.c
index ac3799f739..b4f8844e76 100644
--- a/block/blkdebug.c
+++ b/block/blkdebug.c
@@ -70,6 +70,8 @@ typedef struct BlkdebugSuspendedReq {
      /* IN: initialized in suspend_request() */
      Coroutine *co;
      char *tag;
+
+    /* List entry protected BDRVBlkdebugState::lock */
      QLIST_ENTRY(BlkdebugSuspendedReq) next;
  } BlkdebugSuspendedReq;
  
@@ -100,6 +102,8 @@ typedef struct BlkdebugRule {
              char *tag;
          } suspend;
      } options;
+
+    /* List entries protected BDRVBlkdebugState::lock */
      QLIST_ENTRY(BlkdebugRule) next;
      QSIMPLEQ_ENTRY(BlkdebugRule) active_next;
  } BlkdebugRule;
@@ -249,9 +253,9 @@ static int add_rule(void *opaque, QemuOpts *opts, Error **errp)
      };
  
      /* Add the rule */
-    qemu_mutex_lock(&s->lock);
-    QLIST_INSERT_HEAD(&s->rules[event], rule, next);
-    qemu_mutex_unlock(&s->lock);
+    WITH_QEMU_LOCK_GUARD(&s->lock) {
+        QLIST_INSERT_HEAD(&s->rules[event], rule, next);
+    }
  
      return 0;
  }
@@ -591,33 +595,32 @@ static int rule_check(BlockDriverState *bs, uint64_t offset, uint64_t bytes,
      int error;
      bool immediately;
  
-    qemu_mutex_lock(&s->lock);
-    QSIMPLEQ_FOREACH(rule, &s->active_rules, active_next) {
-        uint64_t inject_offset = rule->options.inject.offset;
-
-        if ((inject_offset == -1 ||
-             (bytes && inject_offset >= offset &&
-              inject_offset < offset + bytes)) &&
-            (rule->options.inject.iotype_mask & (1ull << iotype)))
-        {
-            break;
+    WITH_QEMU_LOCK_GUARD(&s->lock) {
+        QSIMPLEQ_FOREACH(rule, &s->active_rules, active_next) {
+            uint64_t inject_offset = rule->options.inject.offset;
+
+            if ((inject_offset == -1 ||
+                 (bytes && inject_offset >= offset &&
+                  inject_offset < offset + bytes)) &&
+                (rule->options.inject.iotype_mask & (1ull << iotype)))
+            {
+                break;
+            }
          }
-    }
  
-    if (!rule || !rule->options.inject.error) {
-        qemu_mutex_unlock(&s->lock);
-        return 0;
-    }
+        if (!rule || !rule->options.inject.error) {
+            return 0;
+        }
  
-    immediately = rule->options.inject.immediately;
-    error = rule->options.inject.error;
+        immediately = rule->options.inject.immediately;
+        error = rule->options.inject.error;
  
-    if (rule->options.inject.once) {
-        QSIMPLEQ_REMOVE(&s->active_rules, rule, BlkdebugRule, active_next);
-        remove_rule(rule);
+        if (rule->options.inject.once) {
+            QSIMPLEQ_REMOVE(&s->active_rules, rule, BlkdebugRule, active_next);
+            remove_rule(rule);
+        }
      }
  
-    qemu_mutex_unlock(&s->lock);
      if (!immediately) {
          aio_co_schedule(qemu_get_current_aio_context(), qemu_coroutine_self());
          qemu_coroutine_yield();
@@ -880,9 +883,9 @@ static int blkdebug_debug_breakpoint(BlockDriverState *bs, const char *event,
          .options.suspend.tag = g_strdup(tag),
      };
  
-    qemu_mutex_lock(&s->lock);
-    QLIST_INSERT_HEAD(&s->rules[blkdebug_event], rule, next);
-    qemu_mutex_unlock(&s->lock);
+    WITH_QEMU_LOCK_GUARD(&s->lock) {
+        QLIST_INSERT_HEAD(&s->rules[blkdebug_event], rule, next);
+    }
  
      return 0;
  }




-- 
Best regards,
Vladimir