[PATCH 0/6] block-copy: make helper APIs thread safe

[PATCH 0/6] block-copy: make helper APIs thread safe

[Emanuele Giuseppe Esposito]

This serie of patches bring thread safety to the smaller APIs used by
block-copy, namely ratelimit, progressmeter, co-shared-resource
and aiotask.
The end goal is to reduce the usage of the global
AioContexlock in block-copy, by introducing smaller granularity
locks thus on making the block layer thread safe. 

This serie depends on Paolo's coroutine_sleep API.

What's missing for block-copy to be fully thread-safe is fixing
the CoSleep API to allow cross-thread sleep and wakeup.
Paolo is working on it and will post the patches once his new
CoSleep API is accepted.

Patches 1-3 work on ratelimit (they are also based on the first
ratelimit patch sent by Paolo), 4 covers progressmeter,
5 co-shared-resources and 6 aiopool.

Based-on: <20210503112550.478521-1-pbonzini@redhat.com> [coroutine_sleep]
Based-on: <20210413125533.217440-1-pbonzini@redhat.com> [ratelimit]
Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
---
v1 -> v2:
* More field categorized as IN/State/OUT in the various struct
* Fix a couple of places where I missed locks [Vladimir, Paolo]

Emanuele Giuseppe Esposito (3):
  progressmeter: protect with a mutex
  co-shared-resource: protect with a mutex
  aiopool: protect with a mutex

Paolo Bonzini (3):
  ratelimit: treat zero speed as unlimited
  block-copy: let ratelimit handle a speed of 0
  blockjob: let ratelimit handle a speed of 0

 block/aio_task.c               | 63 ++++++++++++++++++++--------------
 block/block-copy.c             | 28 ++++++---------
 blockjob.c                     | 45 +++++++++++++++---------
 include/block/aio_task.h       |  2 +-
 include/qemu/progress_meter.h  | 31 +++++++++++++++++
 include/qemu/ratelimit.h       | 12 +++++--
 job-qmp.c                      |  8 +++--
 job.c                          |  3 ++
 qemu-img.c                     |  9 +++--
 util/qemu-co-shared-resource.c | 26 +++++++++++---
 10 files changed, 155 insertions(+), 72 deletions(-)

-- 
2.30.2

[PATCH 1/6] ratelimit: treat zero speed as unlimited

[Emanuele Giuseppe Esposito]

From: Paolo Bonzini <pbonzini@redhat.com>

Both users of RateLimit, block-copy.c and blockjob.c, treat
a speed of zero as unlimited, while RateLimit treats it as
"as slow as possible".  The latter is nicer from the code
point of view but pretty useless, so disable rate limiting
if a speed of zero is provided.

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
---
 include/qemu/ratelimit.h | 12 ++++++++++--
 1 file changed, 10 insertions(+), 2 deletions(-)

diff --git a/include/qemu/ratelimit.h b/include/qemu/ratelimit.h
index 003ea6d5a3..48bf59e857 100644
--- a/include/qemu/ratelimit.h
+++ b/include/qemu/ratelimit.h
@@ -43,7 +43,11 @@ static inline int64_t ratelimit_calculate_delay(RateLimit *limit, uint64_t n)
     double delay_slices;
 
     QEMU_LOCK_GUARD(&limit->lock);
-    assert(limit->slice_quota && limit->slice_ns);
+    if (!limit->slice_quota) {
+        /* Throttling disabled.  */
+        return 0;
+    }
+    assert(limit->slice_ns);
 
     if (limit->slice_end_time < now) {
         /* Previous, possibly extended, time slice finished; reset the
@@ -83,7 +87,11 @@ static inline void ratelimit_set_speed(RateLimit *limit, uint64_t speed,
 {
     QEMU_LOCK_GUARD(&limit->lock);
     limit->slice_ns = slice_ns;
-    limit->slice_quota = MAX(((double)speed * slice_ns) / 1000000000ULL, 1);
+    if (speed == 0) {
+        limit->slice_quota = 0;
+    } else {
+        limit->slice_quota = MAX(((double)speed * slice_ns) / 1000000000ULL, 1);
+    }
 }
 
 #endif
-- 
2.30.2

[PATCH 2/6] block-copy: let ratelimit handle a speed of 0

[Emanuele Giuseppe Esposito]

From: Paolo Bonzini <pbonzini@redhat.com>

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
---
 block/block-copy.c | 28 ++++++++++------------------
 1 file changed, 10 insertions(+), 18 deletions(-)

diff --git a/block/block-copy.c b/block/block-copy.c
index c2e5090412..7e9467d48a 100644
--- a/block/block-copy.c
+++ b/block/block-copy.c
@@ -113,7 +113,6 @@ typedef struct BlockCopyState {
 
     SharedResource *mem;
 
-    uint64_t speed;
     RateLimit rate_limit;
 } BlockCopyState;
 
@@ -619,23 +618,19 @@ block_copy_dirty_clusters(BlockCopyCallState *call_state)
         }
         task->zeroes = ret & BDRV_BLOCK_ZERO;
 
-        if (s->speed) {
-            if (!call_state->ignore_ratelimit) {
-                uint64_t ns = ratelimit_calculate_delay(&s->rate_limit, 0);
-                if (ns > 0) {
-                    block_copy_task_end(task, -EAGAIN);
-                    g_free(task);
-                    qemu_co_sleep_ns_wakeable(&call_state->sleep,
-                                              QEMU_CLOCK_REALTIME, ns);
-                    continue;
-                }
+        if (!call_state->ignore_ratelimit) {
+            uint64_t ns = ratelimit_calculate_delay(&s->rate_limit, 0);
+            if (ns > 0) {
+                block_copy_task_end(task, -EAGAIN);
+                g_free(task);
+                qemu_co_sleep_ns_wakeable(&call_state->sleep,
+                                            QEMU_CLOCK_REALTIME, ns);
+                continue;
             }
-
-            ratelimit_calculate_delay(&s->rate_limit, task->bytes);
         }
 
+        ratelimit_calculate_delay(&s->rate_limit, task->bytes);
         trace_block_copy_process(s, task->offset);
-
         co_get_from_shres(s->mem, task->bytes);
 
         offset = task_end(task);
@@ -825,10 +820,7 @@ void block_copy_set_skip_unallocated(BlockCopyState *s, bool skip)
 
 void block_copy_set_speed(BlockCopyState *s, uint64_t speed)
 {
-    s->speed = speed;
-    if (speed > 0) {
-        ratelimit_set_speed(&s->rate_limit, speed, BLOCK_COPY_SLICE_TIME);
-    }
+    ratelimit_set_speed(&s->rate_limit, speed, BLOCK_COPY_SLICE_TIME);
 
     /*
      * Note: it's good to kick all call states from here, but it should be done
-- 
2.30.2

[PATCH 3/6] blockjob: let ratelimit handle a speed of 0

[Emanuele Giuseppe Esposito]

From: Paolo Bonzini <pbonzini@redhat.com>

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
---
 blockjob.c | 12 +++---------
 1 file changed, 3 insertions(+), 9 deletions(-)

diff --git a/blockjob.c b/blockjob.c
index dc1d9e0e46..046c1bcd66 100644
--- a/blockjob.c
+++ b/blockjob.c
@@ -300,10 +300,6 @@ bool block_job_set_speed(BlockJob *job, int64_t speed, Error **errp)
 
 int64_t block_job_ratelimit_get_delay(BlockJob *job, uint64_t n)
 {
-    if (!job->speed) {
-        return 0;
-    }
-
     return ratelimit_calculate_delay(&job->limit, n);
 }
 
@@ -473,11 +469,9 @@ void *block_job_create(const char *job_id, const BlockJobDriver *driver,
     blk_set_allow_aio_context_change(blk, true);
 
     /* Only set speed when necessary to avoid NotSupported error */
-    if (speed != 0) {
-        if (!block_job_set_speed(job, speed, errp)) {
-            job_early_fail(&job->job);
-            return NULL;
-        }
+    if (!block_job_set_speed(job, speed, errp)) {
+        job_early_fail(&job->job);
+        return NULL;
     }
 
     return job;
-- 
2.30.2

[PATCH 4/6] progressmeter: protect with a mutex

[Emanuele Giuseppe Esposito]

Progressmeter is protected by the AioContext mutex, which
is taken by the block jobs and their caller (like blockdev).

We would like to remove the dependency of block layer code on the
AioContext mutex, since most drivers and the core I/O code are already
not relying on it.

The simplest thing to do is to add a mutex to be able to provide
an accurate snapshot of the progress values to the caller.

Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
---
 blockjob.c                    | 33 +++++++++++++++++++++++++--------
 include/qemu/progress_meter.h | 31 +++++++++++++++++++++++++++++++
 job-qmp.c                     |  8 ++++++--
 job.c                         |  3 +++
 qemu-img.c                    |  9 ++++++---
 5 files changed, 71 insertions(+), 13 deletions(-)

diff --git a/blockjob.c b/blockjob.c
index 046c1bcd66..98bb12f015 100644
--- a/blockjob.c
+++ b/blockjob.c
@@ -306,18 +306,23 @@ int64_t block_job_ratelimit_get_delay(BlockJob *job, uint64_t n)
 BlockJobInfo *block_job_query(BlockJob *job, Error **errp)
 {
     BlockJobInfo *info;
+    uint64_t progress_current, progress_total;
 
     if (block_job_is_internal(job)) {
         error_setg(errp, "Cannot query QEMU internal jobs");
         return NULL;
     }
+
+    progress_get_snapshot(&job->job.progress, &progress_current,
+                          &progress_total);
+
     info = g_new0(BlockJobInfo, 1);
     info->type      = g_strdup(job_type_str(&job->job));
     info->device    = g_strdup(job->job.id);
     info->busy      = qatomic_read(&job->job.busy);
     info->paused    = job->job.pause_count > 0;
-    info->offset    = job->job.progress.current;
-    info->len       = job->job.progress.total;
+    info->offset    = progress_current;
+    info->len       = progress_total;
     info->speed     = job->speed;
     info->io_status = job->iostatus;
     info->ready     = job_is_ready(&job->job),
@@ -344,15 +349,19 @@ static void block_job_iostatus_set_err(BlockJob *job, int error)
 static void block_job_event_cancelled(Notifier *n, void *opaque)
 {
     BlockJob *job = opaque;
+    uint64_t progress_current, progress_total;
 
     if (block_job_is_internal(job)) {
         return;
     }
 
+    progress_get_snapshot(&job->job.progress, &progress_current,
+                          &progress_total);
+
     qapi_event_send_block_job_cancelled(job_type(&job->job),
                                         job->job.id,
-                                        job->job.progress.total,
-                                        job->job.progress.current,
+                                        progress_total,
+                                        progress_current,
                                         job->speed);
 }
 
@@ -360,6 +369,7 @@ static void block_job_event_completed(Notifier *n, void *opaque)
 {
     BlockJob *job = opaque;
     const char *msg = NULL;
+    uint64_t progress_current, progress_total;
 
     if (block_job_is_internal(job)) {
         return;
@@ -369,10 +379,13 @@ static void block_job_event_completed(Notifier *n, void *opaque)
         msg = error_get_pretty(job->job.err);
     }
 
+    progress_get_snapshot(&job->job.progress, &progress_current,
+                          &progress_total);
+
     qapi_event_send_block_job_completed(job_type(&job->job),
                                         job->job.id,
-                                        job->job.progress.total,
-                                        job->job.progress.current,
+                                        progress_total,
+                                        progress_current,
                                         job->speed,
                                         !!msg,
                                         msg);
@@ -393,15 +406,19 @@ static void block_job_event_pending(Notifier *n, void *opaque)
 static void block_job_event_ready(Notifier *n, void *opaque)
 {
     BlockJob *job = opaque;
+    uint64_t progress_current, progress_total;
 
     if (block_job_is_internal(job)) {
         return;
     }
 
+    progress_get_snapshot(&job->job.progress, &progress_current,
+                          &progress_total);
+
     qapi_event_send_block_job_ready(job_type(&job->job),
                                     job->job.id,
-                                    job->job.progress.total,
-                                    job->job.progress.current,
+                                    progress_total,
+                                    progress_current,
                                     job->speed);
 }
 
diff --git a/include/qemu/progress_meter.h b/include/qemu/progress_meter.h
index 9a23ff071c..81f79770d2 100644
--- a/include/qemu/progress_meter.h
+++ b/include/qemu/progress_meter.h
@@ -27,6 +27,8 @@
 #ifndef QEMU_PROGRESS_METER_H
 #define QEMU_PROGRESS_METER_H
 
+#include "qemu/lockable.h"
+
 typedef struct ProgressMeter {
     /**
      * Current progress. The unit is arbitrary as long as the ratio between
@@ -37,21 +39,50 @@ typedef struct ProgressMeter {
 
     /** Estimated current value at the completion of the process */
     uint64_t total;
+
+    QemuMutex lock;
 } ProgressMeter;
 
+static inline void progress_init(ProgressMeter *pm)
+{
+    qemu_mutex_init(&pm->lock);
+}
+
+static inline void progress_destroy(ProgressMeter *pm)
+{
+    qemu_mutex_destroy(&pm->lock);
+}
+
+static inline void progress_get_snapshot(ProgressMeter *pm,
+                                         uint64_t *current, uint64_t *total)
+{
+    QEMU_LOCK_GUARD(&pm->lock);
+
+    if (current) {
+        *current = pm->current;
+    }
+
+    if (total) {
+        *total = pm->total;
+    }
+}
+
 static inline void progress_work_done(ProgressMeter *pm, uint64_t done)
 {
+    QEMU_LOCK_GUARD(&pm->lock);
     pm->current += done;
 }
 
 static inline void progress_set_remaining(ProgressMeter *pm, uint64_t remaining)
 {
+    QEMU_LOCK_GUARD(&pm->lock);
     pm->total = pm->current + remaining;
 }
 
 static inline void progress_increase_remaining(ProgressMeter *pm,
                                                uint64_t delta)
 {
+    QEMU_LOCK_GUARD(&pm->lock);
     pm->total += delta;
 }
 
diff --git a/job-qmp.c b/job-qmp.c
index 34c4da094f..829a28aa70 100644
--- a/job-qmp.c
+++ b/job-qmp.c
@@ -144,16 +144,20 @@ void qmp_job_dismiss(const char *id, Error **errp)
 static JobInfo *job_query_single(Job *job, Error **errp)
 {
     JobInfo *info;
+    uint64_t progress_current;
+    uint64_t progress_total;
 
     assert(!job_is_internal(job));
+    progress_get_snapshot(&job->progress, &progress_current,
+                          &progress_total);
 
     info = g_new(JobInfo, 1);
     *info = (JobInfo) {
         .id                 = g_strdup(job->id),
         .type               = job_type(job),
         .status             = job->status,
-        .current_progress   = job->progress.current,
-        .total_progress     = job->progress.total,
+        .current_progress   = progress_current,
+        .total_progress     = progress_total,
         .has_error          = !!job->err,
         .error              = job->err ? \
                               g_strdup(error_get_pretty(job->err)) : NULL,
diff --git a/job.c b/job.c
index 4aff13d95a..00b0560052 100644
--- a/job.c
+++ b/job.c
@@ -339,6 +339,8 @@ void *job_create(const char *job_id, const JobDriver *driver, JobTxn *txn,
     job->cb            = cb;
     job->opaque        = opaque;
 
+    progress_init(&job->progress);
+
     notifier_list_init(&job->on_finalize_cancelled);
     notifier_list_init(&job->on_finalize_completed);
     notifier_list_init(&job->on_pending);
@@ -382,6 +384,7 @@ void job_unref(Job *job)
 
         QLIST_REMOVE(job, job_list);
 
+        progress_destroy(&job->progress);
         error_free(job->err);
         g_free(job->id);
         g_free(job);
diff --git a/qemu-img.c b/qemu-img.c
index a5993682aa..7956a89965 100644
--- a/qemu-img.c
+++ b/qemu-img.c
@@ -900,6 +900,7 @@ static void common_block_job_cb(void *opaque, int ret)
 
 static void run_block_job(BlockJob *job, Error **errp)
 {
+    uint64_t progress_current, progress_total;
     AioContext *aio_context = blk_get_aio_context(job->blk);
     int ret = 0;
 
@@ -908,9 +909,11 @@ static void run_block_job(BlockJob *job, Error **errp)
     do {
         float progress = 0.0f;
         aio_poll(aio_context, true);
-        if (job->job.progress.total) {
-            progress = (float)job->job.progress.current /
-                       job->job.progress.total * 100.f;
+
+        progress_get_snapshot(&job->job.progress, &progress_current,
+                              &progress_total);
+        if (progress_total) {
+            progress = (float)progress_current / progress_total * 100.f;
         }
         qemu_progress_print(progress, 0);
     } while (!job_is_ready(&job->job) && !job_is_completed(&job->job));
-- 
2.30.2

[PATCH 5/6] co-shared-resource: protect with a mutex

[Emanuele Giuseppe Esposito]

co-shared-resource is currently not thread-safe, as also reported
in co-shared-resource.h. Add a QemuMutex because co_try_get_from_shres
can also be invoked from non-coroutine context.

Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
---
 util/qemu-co-shared-resource.c | 26 ++++++++++++++++++++++----
 1 file changed, 22 insertions(+), 4 deletions(-)

diff --git a/util/qemu-co-shared-resource.c b/util/qemu-co-shared-resource.c
index 1c83cd9d29..c455d02a1e 100644
--- a/util/qemu-co-shared-resource.c
+++ b/util/qemu-co-shared-resource.c
@@ -32,6 +32,7 @@ struct SharedResource {
     uint64_t available;
 
     CoQueue queue;
+    QemuMutex lock;
 };
 
 SharedResource *shres_create(uint64_t total)
@@ -40,17 +41,23 @@ SharedResource *shres_create(uint64_t total)
 
     s->total = s->available = total;
     qemu_co_queue_init(&s->queue);
+    qemu_mutex_init(&s->lock);
 
     return s;
 }
 
 void shres_destroy(SharedResource *s)
 {
-    assert(s->available == s->total);
+    WITH_QEMU_LOCK_GUARD(&s->lock) {
+        assert(s->available == s->total);
+    }
+
+    qemu_mutex_destroy(&s->lock);
     g_free(s);
 }
 
-bool co_try_get_from_shres(SharedResource *s, uint64_t n)
+/* Called with lock held */
+static bool co_try_get_from_shres_unlocked(SharedResource *s, uint64_t n)
 {
     if (s->available >= n) {
         s->available -= n;
@@ -60,16 +67,27 @@ bool co_try_get_from_shres(SharedResource *s, uint64_t n)
     return false;
 }
 
+bool co_try_get_from_shres(SharedResource *s, uint64_t n)
+{
+    bool res;
+    QEMU_LOCK_GUARD(&s->lock);
+    res = co_try_get_from_shres_unlocked(s, n);
+
+    return res;
+}
+
 void coroutine_fn co_get_from_shres(SharedResource *s, uint64_t n)
 {
+    QEMU_LOCK_GUARD(&s->lock);
     assert(n <= s->total);
-    while (!co_try_get_from_shres(s, n)) {
-        qemu_co_queue_wait(&s->queue, NULL);
+    while (!co_try_get_from_shres_unlocked(s, n)) {
+        qemu_co_queue_wait(&s->queue, &s->lock);
     }
 }
 
 void coroutine_fn co_put_to_shres(SharedResource *s, uint64_t n)
 {
+    QEMU_LOCK_GUARD(&s->lock);
     assert(s->total - s->available >= n);
     s->available += n;
     qemu_co_queue_restart_all(&s->queue);
-- 
2.30.2

[PATCH 6/6] aiopool: protect with a mutex

[Emanuele Giuseppe Esposito]

Divide the fields in AioTaskPool in IN and Status, and
introduce a CoQueue instead of .wait to take care of suspending
and resuming the calling coroutine, and a lock to protect the
busy_tasks counter accesses and the AioTask .ret field.

Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
---
 block/aio_task.c         | 63 ++++++++++++++++++++++++----------------
 include/block/aio_task.h |  2 +-
 2 files changed, 39 insertions(+), 26 deletions(-)

diff --git a/block/aio_task.c b/block/aio_task.c
index 88989fa248..7ac6b5dd72 100644
--- a/block/aio_task.c
+++ b/block/aio_task.c
@@ -27,62 +27,70 @@
 #include "block/aio_task.h"
 
 struct AioTaskPool {
-    Coroutine *main_co;
-    int status;
+    /* IN: just set in aio_task_pool_new and never modified */
     int max_busy_tasks;
+
+    /* Status: either atomic or protected by the lock */
+    int status;
     int busy_tasks;
-    bool waiting;
+    CoQueue queue;
+    CoMutex lock;
 };
 
 static void coroutine_fn aio_task_co(void *opaque)
 {
+    int ret;
     AioTask *task = opaque;
     AioTaskPool *pool = task->pool;
 
-    assert(pool->busy_tasks < pool->max_busy_tasks);
-    pool->busy_tasks++;
+    WITH_QEMU_LOCK_GUARD(&pool->lock) {
+        assert(pool->busy_tasks < pool->max_busy_tasks);
+        pool->busy_tasks++;
 
-    task->ret = task->func(task);
+        ret = task->func(task);
+        task->ret = ret;
 
-    pool->busy_tasks--;
+        pool->busy_tasks--;
+    }
 
-    if (task->ret < 0 && pool->status == 0) {
-        pool->status = task->ret;
+    if (ret < 0) {
+        qatomic_cmpxchg(&pool->status, 0, ret);
     }
 
     g_free(task);
 
-    if (pool->waiting) {
-        pool->waiting = false;
-        aio_co_wake(pool->main_co);
-    }
+    qemu_co_queue_next(&pool->queue);
 }
 
-void coroutine_fn aio_task_pool_wait_one(AioTaskPool *pool)
+/* Called with lock held */
+static void coroutine_fn aio_task_pool_wait_one_unlocked(AioTaskPool *pool)
 {
     assert(pool->busy_tasks > 0);
-    assert(qemu_coroutine_self() == pool->main_co);
-
-    pool->waiting = true;
-    qemu_coroutine_yield();
-
-    assert(!pool->waiting);
+    qemu_co_queue_wait(&pool->queue, &pool->lock);
     assert(pool->busy_tasks < pool->max_busy_tasks);
 }
 
+void coroutine_fn aio_task_pool_wait_one(AioTaskPool *pool)
+{
+    QEMU_LOCK_GUARD(&pool->lock);
+    aio_task_pool_wait_one_unlocked(pool);
+}
+
 void coroutine_fn aio_task_pool_wait_slot(AioTaskPool *pool)
 {
+    QEMU_LOCK_GUARD(&pool->lock);
     if (pool->busy_tasks < pool->max_busy_tasks) {
         return;
     }
 
-    aio_task_pool_wait_one(pool);
+    aio_task_pool_wait_one_unlocked(pool);
 }
 
 void coroutine_fn aio_task_pool_wait_all(AioTaskPool *pool)
 {
+    QEMU_LOCK_GUARD(&pool->lock);
     while (pool->busy_tasks > 0) {
-        aio_task_pool_wait_one(pool);
+        aio_task_pool_wait_one_unlocked(pool);
     }
 }
 
@@ -98,8 +106,8 @@ AioTaskPool *coroutine_fn aio_task_pool_new(int max_busy_tasks)
 {
     AioTaskPool *pool = g_new0(AioTaskPool, 1);
 
-    pool->main_co = qemu_coroutine_self();
     pool->max_busy_tasks = max_busy_tasks;
+    qemu_co_queue_init(&pool->queue);
 
     return pool;
 }
@@ -115,10 +123,15 @@ int aio_task_pool_status(AioTaskPool *pool)
         return 0; /* Sugar for lazy allocation of aio pool */
     }
 
-    return pool->status;
+    return qatomic_read(&pool->status);
 }
 
 bool aio_task_pool_empty(AioTaskPool *pool)
 {
-    return pool->busy_tasks == 0;
+    int tasks;
+
+    qemu_co_mutex_lock(&pool->lock);
+    tasks = pool->busy_tasks;
+    qemu_co_mutex_unlock(&pool->lock);
+    return tasks == 0;
 }
diff --git a/include/block/aio_task.h b/include/block/aio_task.h
index 50bc1e1817..b22a4310aa 100644
--- a/include/block/aio_task.h
+++ b/include/block/aio_task.h
@@ -33,7 +33,7 @@ typedef int coroutine_fn (*AioTaskFunc)(AioTask *task);
 struct AioTask {
     AioTaskPool *pool;
     AioTaskFunc func;
-    int ret;
+    int ret; /* atomic */
 };
 
 AioTaskPool *coroutine_fn aio_task_pool_new(int max_busy_tasks);
-- 
2.30.2

Re: [PATCH 0/6] block-copy: make helper APIs thread safe

[Vladimir Sementsov-Ogievskiy]

10.05.2021 11:59, Emanuele Giuseppe Esposito wrote:
> This serie of patches bring thread safety to the smaller APIs used by
> block-copy, namely ratelimit, progressmeter, co-shared-resource
> and aiotask.
> The end goal is to reduce the usage of the global
> AioContexlock in block-copy, by introducing smaller granularity
> locks thus on making the block layer thread safe.
> 
> This serie depends on Paolo's coroutine_sleep API.
> 
> What's missing for block-copy to be fully thread-safe is fixing
> the CoSleep API to allow cross-thread sleep and wakeup.
> Paolo is working on it and will post the patches once his new
> CoSleep API is accepted.
> 
> Patches 1-3 work on ratelimit (they are also based on the first
> ratelimit patch sent by Paolo), 4 covers progressmeter,
> 5 co-shared-resources and 6 aiopool.
> 
> Based-on: <20210503112550.478521-1-pbonzini@redhat.com> [coroutine_sleep]
> Based-on: <20210413125533.217440-1-pbonzini@redhat.com> [ratelimit]

Seems, patchew failed to parse your "Based-on" tags: https://patchew.org/QEMU/20210510085941.22769-1-eesposit@redhat.com/
(and tries to apply series on master)
I think, it's because you placed "[...]" comments on the same lines.

Interesting, will patchew see, if I just duplicate tags here:

Based-on: <20210503112550.478521-1-pbonzini@redhat.com>
Based-on: <20210413125533.217440-1-pbonzini@redhat.com>


> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
> ---
> v1 -> v2:
> * More field categorized as IN/State/OUT in the various struct
> * Fix a couple of places where I missed locks [Vladimir, Paolo]
> 
> Emanuele Giuseppe Esposito (3):
>    progressmeter: protect with a mutex
>    co-shared-resource: protect with a mutex
>    aiopool: protect with a mutex
> 
> Paolo Bonzini (3):
>    ratelimit: treat zero speed as unlimited
>    block-copy: let ratelimit handle a speed of 0
>    blockjob: let ratelimit handle a speed of 0
> 
>   block/aio_task.c               | 63 ++++++++++++++++++++--------------
>   block/block-copy.c             | 28 ++++++---------
>   blockjob.c                     | 45 +++++++++++++++---------
>   include/block/aio_task.h       |  2 +-
>   include/qemu/progress_meter.h  | 31 +++++++++++++++++
>   include/qemu/ratelimit.h       | 12 +++++--
>   job-qmp.c                      |  8 +++--
>   job.c                          |  3 ++
>   qemu-img.c                     |  9 +++--
>   util/qemu-co-shared-resource.c | 26 +++++++++++---
>   10 files changed, 155 insertions(+), 72 deletions(-)
> 


-- 
Best regards,
Vladimir

Re: [PATCH 1/6] ratelimit: treat zero speed as unlimited

[Vladimir Sementsov-Ogievskiy]

10.05.2021 11:59, Emanuele Giuseppe Esposito wrote:
> From: Paolo Bonzini <pbonzini@redhat.com>
> 
> Both users of RateLimit, block-copy.c and blockjob.c, treat
> a speed of zero as unlimited, while RateLimit treats it as
> "as slow as possible".  The latter is nicer from the code
> point of view but pretty useless, so disable rate limiting
> if a speed of zero is provided.
> 
> Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>

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

> ---
>   include/qemu/ratelimit.h | 12 ++++++++++--
>   1 file changed, 10 insertions(+), 2 deletions(-)
> 
> diff --git a/include/qemu/ratelimit.h b/include/qemu/ratelimit.h
> index 003ea6d5a3..48bf59e857 100644
> --- a/include/qemu/ratelimit.h
> +++ b/include/qemu/ratelimit.h
> @@ -43,7 +43,11 @@ static inline int64_t ratelimit_calculate_delay(RateLimit *limit, uint64_t n)
>       double delay_slices;
>   
>       QEMU_LOCK_GUARD(&limit->lock);
> -    assert(limit->slice_quota && limit->slice_ns);
> +    if (!limit->slice_quota) {
> +        /* Throttling disabled.  */
> +        return 0;
> +    }
> +    assert(limit->slice_ns);
>   
>       if (limit->slice_end_time < now) {
>           /* Previous, possibly extended, time slice finished; reset the
> @@ -83,7 +87,11 @@ static inline void ratelimit_set_speed(RateLimit *limit, uint64_t speed,
>   {
>       QEMU_LOCK_GUARD(&limit->lock);
>       limit->slice_ns = slice_ns;
> -    limit->slice_quota = MAX(((double)speed * slice_ns) / 1000000000ULL, 1);
> +    if (speed == 0) {
> +        limit->slice_quota = 0;
> +    } else {
> +        limit->slice_quota = MAX(((double)speed * slice_ns) / 1000000000ULL, 1);
> +    }
>   }
>   
>   #endif
> 


-- 
Best regards,
Vladimir

Re: [PATCH 2/6] block-copy: let ratelimit handle a speed of 0

[Vladimir Sementsov-Ogievskiy]

10.05.2021 11:59, Emanuele Giuseppe Esposito wrote:
> From: Paolo Bonzini <pbonzini@redhat.com>
> 
> Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
> ---
>   block/block-copy.c | 28 ++++++++++------------------
>   1 file changed, 10 insertions(+), 18 deletions(-)
> 
> diff --git a/block/block-copy.c b/block/block-copy.c
> index c2e5090412..7e9467d48a 100644
> --- a/block/block-copy.c
> +++ b/block/block-copy.c
> @@ -113,7 +113,6 @@ typedef struct BlockCopyState {
>   
>       SharedResource *mem;
>   
> -    uint64_t speed;
>       RateLimit rate_limit;
>   } BlockCopyState;
>   
> @@ -619,23 +618,19 @@ block_copy_dirty_clusters(BlockCopyCallState *call_state)
>           }
>           task->zeroes = ret & BDRV_BLOCK_ZERO;
>   
> -        if (s->speed) {
> -            if (!call_state->ignore_ratelimit) {
> -                uint64_t ns = ratelimit_calculate_delay(&s->rate_limit, 0);
> -                if (ns > 0) {
> -                    block_copy_task_end(task, -EAGAIN);
> -                    g_free(task);
> -                    qemu_co_sleep_ns_wakeable(&call_state->sleep,
> -                                              QEMU_CLOCK_REALTIME, ns);
> -                    continue;
> -                }
> +        if (!call_state->ignore_ratelimit) {
> +            uint64_t ns = ratelimit_calculate_delay(&s->rate_limit, 0);
> +            if (ns > 0) {
> +                block_copy_task_end(task, -EAGAIN);
> +                g_free(task);
> +                qemu_co_sleep_ns_wakeable(&call_state->sleep,
> +                                            QEMU_CLOCK_REALTIME, ns);

indentation broken

> +                continue;
>               }
> -
> -            ratelimit_calculate_delay(&s->rate_limit, task->bytes);
>           }
>   
> +        ratelimit_calculate_delay(&s->rate_limit, task->bytes);
>           trace_block_copy_process(s, task->offset);
> -

I'd keep both newlines around trace_, as all three calls has no relation to each other..

>           co_get_from_shres(s->mem, task->bytes);
>   
>           offset = task_end(task);
> @@ -825,10 +820,7 @@ void block_copy_set_skip_unallocated(BlockCopyState *s, bool skip)
>   
>   void block_copy_set_speed(BlockCopyState *s, uint64_t speed)
>   {
> -    s->speed = speed;
> -    if (speed > 0) {
> -        ratelimit_set_speed(&s->rate_limit, speed, BLOCK_COPY_SLICE_TIME);
> -    }
> +    ratelimit_set_speed(&s->rate_limit, speed, BLOCK_COPY_SLICE_TIME);
>   
>       /*
>        * Note: it's good to kick all call states from here, but it should be done
> 

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

-- 
Best regards,
Vladimir

Re: [PATCH 3/6] blockjob: let ratelimit handle a speed of 0

[Vladimir Sementsov-Ogievskiy]

10.05.2021 11:59, Emanuele Giuseppe Esposito wrote:
> From: Paolo Bonzini <pbonzini@redhat.com>
> 
> Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
> ---
>   blockjob.c | 12 +++---------
>   1 file changed, 3 insertions(+), 9 deletions(-)
> 
> diff --git a/blockjob.c b/blockjob.c
> index dc1d9e0e46..046c1bcd66 100644
> --- a/blockjob.c
> +++ b/blockjob.c
> @@ -300,10 +300,6 @@ bool block_job_set_speed(BlockJob *job, int64_t speed, Error **errp)
>   
>   int64_t block_job_ratelimit_get_delay(BlockJob *job, uint64_t n)
>   {
> -    if (!job->speed) {
> -        return 0;
> -    }
> -
>       return ratelimit_calculate_delay(&job->limit, n);
>   }
>   
> @@ -473,11 +469,9 @@ void *block_job_create(const char *job_id, const BlockJobDriver *driver,
>       blk_set_allow_aio_context_change(blk, true);
>   
>       /* Only set speed when necessary to avoid NotSupported error */

strange comment, I'd drop it.

> -    if (speed != 0) {
> -        if (!block_job_set_speed(job, speed, errp)) {
> -            job_early_fail(&job->job);
> -            return NULL;
> -        }
> +    if (!block_job_set_speed(job, speed, errp)) {
> +        job_early_fail(&job->job);
> +        return NULL;
>       }
>   
>       return job;
> 

Probably, we should store speed into RateLimit, and add ratelimit_get_speed() call. Then, drop job->speed field.


with dropped "Only set" comment:
Reviewed-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>

-- 
Best regards,
Vladimir

Re: [PATCH 4/6] progressmeter: protect with a mutex

[Vladimir Sementsov-Ogievskiy]

10.05.2021 11:59, Emanuele Giuseppe Esposito wrote:
> Progressmeter is protected by the AioContext mutex, which
> is taken by the block jobs and their caller (like blockdev).
> 
> We would like to remove the dependency of block layer code on the
> AioContext mutex, since most drivers and the core I/O code are already
> not relying on it.
> 
> The simplest thing to do is to add a mutex to be able to provide
> an accurate snapshot of the progress values to the caller.
> 
> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
> ---
>   blockjob.c                    | 33 +++++++++++++++++++++++++--------

[..]

> --- a/include/qemu/progress_meter.h
> +++ b/include/qemu/progress_meter.h
> @@ -27,6 +27,8 @@
>   #ifndef QEMU_PROGRESS_METER_H
>   #define QEMU_PROGRESS_METER_H
>   
> +#include "qemu/lockable.h"
> +
>   typedef struct ProgressMeter {
>       /**
>        * Current progress. The unit is arbitrary as long as the ratio between
> @@ -37,21 +39,50 @@ typedef struct ProgressMeter {
>   
>       /** Estimated current value at the completion of the process */
>       uint64_t total;
> +
> +    QemuMutex lock;
>   } ProgressMeter;
>   
> +static inline void progress_init(ProgressMeter *pm)
> +{
> +    qemu_mutex_init(&pm->lock);
> +}
> +
> +static inline void progress_destroy(ProgressMeter *pm)
> +{
> +    qemu_mutex_destroy(&pm->lock);
> +}


Could we instead add a c file and add the structure private? Then we'll have progress_new() and progress_free() APIs instead.

This way, it would be a lot simpler to control that nobady use structure fields directly.



> +
> +static inline void progress_get_snapshot(ProgressMeter *pm,
> +                                         uint64_t *current, uint64_t *total)
> +{
> +    QEMU_LOCK_GUARD(&pm->lock);
> +
> +    if (current) {
> +        *current = pm->current;
> +    }
> +
> +    if (total) {
> +        *total = pm->total;
> +    }
> +}

We don't have caller that pass only one pointer. So we can just do

*current = pm->current;
*total = pm->total;

implicitly requiring both pointers to be non NULL.



-- 
Best regards,
Vladimir

Re: [PATCH 5/6] co-shared-resource: protect with a mutex

[Vladimir Sementsov-Ogievskiy]

10.05.2021 11:59, Emanuele Giuseppe Esposito wrote:
> co-shared-resource is currently not thread-safe, as also reported
> in co-shared-resource.h. Add a QemuMutex because co_try_get_from_shres
> can also be invoked from non-coroutine context.

But it doesn't. It's called only from co_get_from_shres(). So, better make it a static function first.

> 
> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
> ---
>   util/qemu-co-shared-resource.c | 26 ++++++++++++++++++++++----
>   1 file changed, 22 insertions(+), 4 deletions(-)
> 
> diff --git a/util/qemu-co-shared-resource.c b/util/qemu-co-shared-resource.c
> index 1c83cd9d29..c455d02a1e 100644
> --- a/util/qemu-co-shared-resource.c
> +++ b/util/qemu-co-shared-resource.c
> @@ -32,6 +32,7 @@ struct SharedResource {
>       uint64_t available;
>   
>       CoQueue queue;
> +    QemuMutex lock;
>   };
>   
>   SharedResource *shres_create(uint64_t total)
> @@ -40,17 +41,23 @@ SharedResource *shres_create(uint64_t total)
>   
>       s->total = s->available = total;
>       qemu_co_queue_init(&s->queue);
> +    qemu_mutex_init(&s->lock);
>   
>       return s;
>   }
>   
>   void shres_destroy(SharedResource *s)
>   {
> -    assert(s->available == s->total);
> +    WITH_QEMU_LOCK_GUARD(&s->lock) {
> +        assert(s->available == s->total);
> +    }

shres_destroy can't be thread-safe anyway, as it does qemu_mutex_destroy() and g_free. So, let's don't try to make part of shres_destroy() "thread safe".

> +
> +    qemu_mutex_destroy(&s->lock);
>       g_free(s);
>   }
>   
> -bool co_try_get_from_shres(SharedResource *s, uint64_t n)
> +/* Called with lock held */

it should be called _locked() than.


> +static bool co_try_get_from_shres_unlocked(SharedResource *s, uint64_t n)>   {
>       if (s->available >= n) {
>           s->available -= n;
> @@ -60,16 +67,27 @@ bool co_try_get_from_shres(SharedResource *s, uint64_t n)
>       return false;
>   }
>   
> +bool co_try_get_from_shres(SharedResource *s, uint64_t n)
> +{
> +    bool res;
> +    QEMU_LOCK_GUARD(&s->lock);
> +    res = co_try_get_from_shres_unlocked(s, n);
> +
> +    return res;
> +}
> +
>   void coroutine_fn co_get_from_shres(SharedResource *s, uint64_t n)
>   {
> +    QEMU_LOCK_GUARD(&s->lock);
>       assert(n <= s->total);
> -    while (!co_try_get_from_shres(s, n)) {
> -        qemu_co_queue_wait(&s->queue, NULL);
> +    while (!co_try_get_from_shres_unlocked(s, n)) {
> +        qemu_co_queue_wait(&s->queue, &s->lock);
>       }
>   }
>   
>   void coroutine_fn co_put_to_shres(SharedResource *s, uint64_t n)
>   {
> +    QEMU_LOCK_GUARD(&s->lock);
>       assert(s->total - s->available >= n);
>       s->available += n;
>       qemu_co_queue_restart_all(&s->queue);
> 


-- 
Best regards,
Vladimir

Re: [PATCH 6/6] aiopool: protect with a mutex

[Vladimir Sementsov-Ogievskiy]

10.05.2021 11:59, Emanuele Giuseppe Esposito wrote:
> Divide the fields in AioTaskPool in IN and Status, and
> introduce a CoQueue instead of .wait to take care of suspending
> and resuming the calling coroutine, and a lock to protect the
> busy_tasks counter accesses and the AioTask .ret field.

"and" in commit message is almost always a sign, that there should be several commits :)

Please, do at least refactoring to drop "main_co" in separate of adding a mutex.

Hmm, actually, that was done in Denis's series "[PATCH v8 0/6] block: seriously improve savevm/loadvm performance". (https://patchew.org/QEMU/20200709132644.28470-1-den@openvz.org/)

Probably, you could reuse patches 01,02 of it.

(add Den to cc)

> 
> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
> ---
>   block/aio_task.c         | 63 ++++++++++++++++++++++++----------------
>   include/block/aio_task.h |  2 +-
>   2 files changed, 39 insertions(+), 26 deletions(-)
> 
> diff --git a/block/aio_task.c b/block/aio_task.c
> index 88989fa248..7ac6b5dd72 100644
> --- a/block/aio_task.c
> +++ b/block/aio_task.c
> @@ -27,62 +27,70 @@
>   #include "block/aio_task.h"
>   
>   struct AioTaskPool {
> -    Coroutine *main_co;
> -    int status;
> +    /* IN: just set in aio_task_pool_new and never modified */
>       int max_busy_tasks;
> +
> +    /* Status: either atomic or protected by the lock */
> +    int status;
>       int busy_tasks;
> -    bool waiting;
> +    CoQueue queue;
> +    CoMutex lock;
>   };
>   
>   static void coroutine_fn aio_task_co(void *opaque)
>   {
> +    int ret;
>       AioTask *task = opaque;
>       AioTaskPool *pool = task->pool;
>   
> -    assert(pool->busy_tasks < pool->max_busy_tasks);
> -    pool->busy_tasks++;
> +    WITH_QEMU_LOCK_GUARD(&pool->lock) {
> +        assert(pool->busy_tasks < pool->max_busy_tasks);
> +        pool->busy_tasks++;
>   
> -    task->ret = task->func(task);
> +        ret = task->func(task);
> +        task->ret = ret;
>   
> -    pool->busy_tasks--;
> +        pool->busy_tasks--;
> +    }
>   
> -    if (task->ret < 0 && pool->status == 0) {
> -        pool->status = task->ret;
> +    if (ret < 0) {
> +        qatomic_cmpxchg(&pool->status, 0, ret);
>       }

Can we just do it inside critical section above and avoid extra cmpxchg? We'll need just qatomic_set as a pair to qatomic_read()

>   
>       g_free(task);
>   
> -    if (pool->waiting) {
> -        pool->waiting = false;
> -        aio_co_wake(pool->main_co);
> -    }
> +    qemu_co_queue_next(&pool->queue);

this call doesn't need mutex protection? Then we should modify comment insid AioTaskPool structure.

Anyway, I think it's simpler to just have one QEMU_MUTEX_GUARD() for the whole function.

>   }
>   
> -void coroutine_fn aio_task_pool_wait_one(AioTaskPool *pool)
> +/* Called with lock held */

again, such things usually called _locked().

> +static void coroutine_fn aio_task_pool_wait_one_unlocked(AioTaskPool *pool)
>   {
>       assert(pool->busy_tasks > 0);
> -    assert(qemu_coroutine_self() == pool->main_co);
> -
> -    pool->waiting = true;
> -    qemu_coroutine_yield();
> -
> -    assert(!pool->waiting);
> +    qemu_co_queue_wait(&pool->queue, &pool->lock);
>       assert(pool->busy_tasks < pool->max_busy_tasks);
>   }
>   
> +void coroutine_fn aio_task_pool_wait_one(AioTaskPool *pool)
> +{
> +    QEMU_LOCK_GUARD(&pool->lock);
> +    aio_task_pool_wait_one_unlocked(pool);
> +}
> +
>   void coroutine_fn aio_task_pool_wait_slot(AioTaskPool *pool)
>   {
> +    QEMU_LOCK_GUARD(&pool->lock);
>       if (pool->busy_tasks < pool->max_busy_tasks) {
>           return;
>       }
>   
> -    aio_task_pool_wait_one(pool);
> +    aio_task_pool_wait_one_unlocked(pool);
>   }
>   
>   void coroutine_fn aio_task_pool_wait_all(AioTaskPool *pool)
>   {
> +    QEMU_LOCK_GUARD(&pool->lock);
>       while (pool->busy_tasks > 0) {
> -        aio_task_pool_wait_one(pool);
> +        aio_task_pool_wait_one_unlocked(pool);
>       }
>   }
>   
> @@ -98,8 +106,8 @@ AioTaskPool *coroutine_fn aio_task_pool_new(int max_busy_tasks)
>   {
>       AioTaskPool *pool = g_new0(AioTaskPool, 1);
>   
> -    pool->main_co = qemu_coroutine_self();
>       pool->max_busy_tasks = max_busy_tasks;
> +    qemu_co_queue_init(&pool->queue);
>   
>       return pool;
>   }
> @@ -115,10 +123,15 @@ int aio_task_pool_status(AioTaskPool *pool)
>           return 0; /* Sugar for lazy allocation of aio pool */
>       }
>   
> -    return pool->status;
> +    return qatomic_read(&pool->status);
>   }
>   
>   bool aio_task_pool_empty(AioTaskPool *pool)
>   {
> -    return pool->busy_tasks == 0;
> +    int tasks;
> +
> +    qemu_co_mutex_lock(&pool->lock);
> +    tasks = pool->busy_tasks;
> +    qemu_co_mutex_unlock(&pool->lock);
> +    return tasks == 0;
>   }
> diff --git a/include/block/aio_task.h b/include/block/aio_task.h
> index 50bc1e1817..b22a4310aa 100644
> --- a/include/block/aio_task.h
> +++ b/include/block/aio_task.h
> @@ -33,7 +33,7 @@ typedef int coroutine_fn (*AioTaskFunc)(AioTask *task);
>   struct AioTask {
>       AioTaskPool *pool;
>       AioTaskFunc func;
> -    int ret;
> +    int ret; /* atomic */
>   };
>   
>   AioTaskPool *coroutine_fn aio_task_pool_new(int max_busy_tasks);
> 


-- 
Best regards,
Vladimir

Re: [PATCH 4/6] progressmeter: protect with a mutex

[Emanuele Giuseppe Esposito]

On 10/05/2021 13:28, Vladimir Sementsov-Ogievskiy wrote:
> 10.05.2021 11:59, Emanuele Giuseppe Esposito wrote:
>> Progressmeter is protected by the AioContext mutex, which
>> is taken by the block jobs and their caller (like blockdev).
>>
>> We would like to remove the dependency of block layer code on the
>> AioContext mutex, since most drivers and the core I/O code are already
>> not relying on it.
>>
>> The simplest thing to do is to add a mutex to be able to provide
>> an accurate snapshot of the progress values to the caller.
>>
>> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
>> ---
>>   blockjob.c                    | 33 +++++++++++++++++++++++++--------
> 
> [..]
> 
>> --- a/include/qemu/progress_meter.h
>> +++ b/include/qemu/progress_meter.h
>> @@ -27,6 +27,8 @@
>>   #ifndef QEMU_PROGRESS_METER_H
>>   #define QEMU_PROGRESS_METER_H
>> +#include "qemu/lockable.h"
>> +
>>   typedef struct ProgressMeter {
>>       /**
>>        * Current progress. The unit is arbitrary as long as the ratio 
>> between
>> @@ -37,21 +39,50 @@ typedef struct ProgressMeter {
>>       /** Estimated current value at the completion of the process */
>>       uint64_t total;
>> +
>> +    QemuMutex lock;
>>   } ProgressMeter;
>> +static inline void progress_init(ProgressMeter *pm)
>> +{
>> +    qemu_mutex_init(&pm->lock);
>> +}
>> +
>> +static inline void progress_destroy(ProgressMeter *pm)
>> +{
>> +    qemu_mutex_destroy(&pm->lock);
>> +}
> 
> 
> Could we instead add a c file and add the structure private? Then we'll 
> have progress_new() and progress_free() APIs instead.
> 
> This way, it would be a lot simpler to control that nobady use structure 
> fields directly.
> 
Makes sense, I'll do it.
> 
> 
>> +
>> +static inline void progress_get_snapshot(ProgressMeter *pm,
>> +                                         uint64_t *current, uint64_t 
>> *total)
>> +{
>> +    QEMU_LOCK_GUARD(&pm->lock);
>> +
>> +    if (current) {
>> +        *current = pm->current;
>> +    }
>> +
>> +    if (total) {
>> +        *total = pm->total;
>> +    }
>> +}
> 
> We don't have caller that pass only one pointer. So we can just do
> 
> *current = pm->current;
> *total = pm->total;
> 
> implicitly requiring both pointers to be non NULL.

Is it so performance critical that we need to skip these safety checks?
IMHO we can keep it as it is.

Thank you,
Emanuele
> 
> 
> 

Re: [PATCH 4/6] progressmeter: protect with a mutex

[Vladimir Sementsov-Ogievskiy]

10.05.2021 19:52, Emanuele Giuseppe Esposito wrote:
> 
> 
> On 10/05/2021 13:28, Vladimir Sementsov-Ogievskiy wrote:
>> 10.05.2021 11:59, Emanuele Giuseppe Esposito wrote:
>>> Progressmeter is protected by the AioContext mutex, which
>>> is taken by the block jobs and their caller (like blockdev).
>>>
>>> We would like to remove the dependency of block layer code on the
>>> AioContext mutex, since most drivers and the core I/O code are already
>>> not relying on it.
>>>
>>> The simplest thing to do is to add a mutex to be able to provide
>>> an accurate snapshot of the progress values to the caller.
>>>
>>> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
>>> ---
>>>   blockjob.c                    | 33 +++++++++++++++++++++++++--------
>>
>> [..]
>>
>>> --- a/include/qemu/progress_meter.h
>>> +++ b/include/qemu/progress_meter.h
>>> @@ -27,6 +27,8 @@
>>>   #ifndef QEMU_PROGRESS_METER_H
>>>   #define QEMU_PROGRESS_METER_H
>>> +#include "qemu/lockable.h"
>>> +
>>>   typedef struct ProgressMeter {
>>>       /**
>>>        * Current progress. The unit is arbitrary as long as the ratio between
>>> @@ -37,21 +39,50 @@ typedef struct ProgressMeter {
>>>       /** Estimated current value at the completion of the process */
>>>       uint64_t total;
>>> +
>>> +    QemuMutex lock;
>>>   } ProgressMeter;
>>> +static inline void progress_init(ProgressMeter *pm)
>>> +{
>>> +    qemu_mutex_init(&pm->lock);
>>> +}
>>> +
>>> +static inline void progress_destroy(ProgressMeter *pm)
>>> +{
>>> +    qemu_mutex_destroy(&pm->lock);
>>> +}
>>
>>
>> Could we instead add a c file and add the structure private? Then we'll have progress_new() and progress_free() APIs instead.
>>
>> This way, it would be a lot simpler to control that nobady use structure fields directly.
>>
> Makes sense, I'll do it.
>>
>>
>>> +
>>> +static inline void progress_get_snapshot(ProgressMeter *pm,
>>> +                                         uint64_t *current, uint64_t *total)
>>> +{
>>> +    QEMU_LOCK_GUARD(&pm->lock);
>>> +
>>> +    if (current) {
>>> +        *current = pm->current;
>>> +    }
>>> +
>>> +    if (total) {
>>> +        *total = pm->total;
>>> +    }
>>> +}
>>
>> We don't have caller that pass only one pointer. So we can just do
>>
>> *current = pm->current;
>> *total = pm->total;
>>
>> implicitly requiring both pointers to be non NULL.
> 
> Is it so performance critical that we need to skip these safety checks?
> IMHO we can keep it as it is.


Not performance. It's just less code. You propose more complex interface (pointers may be valid pointers or NULL), implemented by more complex code (extra if blocks). And there no users for this. So, I don't see any reason for extra logic and code lines.

What kind of safety you want? All current callers pass non-zero pointers, it's obvious. If someone will create new call, he should look first at function itself, not blindly pass NULL pointers. And if not, he will get clean crash on zero pointer dereference.


-- 
Best regards,
Vladimir

Re: [PATCH 4/6] progressmeter: protect with a mutex

[Emanuele Giuseppe Esposito]

>>>
>>> We don't have caller that pass only one pointer. So we can just do
>>>
>>> *current = pm->current;
>>> *total = pm->total;
>>>
>>> implicitly requiring both pointers to be non NULL.
>>
>> Is it so performance critical that we need to skip these safety checks?
>> IMHO we can keep it as it is.
> 
> 
> Not performance. It's just less code. You propose more complex interface 
> (pointers may be valid pointers or NULL), implemented by more complex 
> code (extra if blocks). And there no users for this. So, I don't see any 
> reason for extra logic and code lines.
> 
> What kind of safety you want? All current callers pass non-zero 
> pointers, it's obvious. If someone will create new call, he should look 
> first at function itself, not blindly pass NULL pointers. And if not, he 
> will get clean crash on zero pointer dereference.

Ok, makes sense. Will remove the ifs.

Thank you,
Emanuele

Re: [PATCH 6/6] aiopool: protect with a mutex

[Paolo Bonzini]

On 10/05/21 13:56, Vladimir Sementsov-Ogievskiy wrote:
>>
>> +    }
>> -    if (task->ret < 0 && pool->status == 0) {
>> -        pool->status = task->ret;
>> +    if (ret < 0) {
>> +        qatomic_cmpxchg(&pool->status, 0, ret);
>>       }
> 
> Can we just do it inside critical section above and avoid extra cmpxchg? 
> We'll need just qatomic_set as a pair to qatomic_read()

Good idea.

>>       g_free(task);
>> -    if (pool->waiting) {
>> -        pool->waiting = false;
>> -        aio_co_wake(pool->main_co);
>> -    }
>> +    qemu_co_queue_next(&pool->queue);
> 
> this call doesn't need mutex protection?

It does indeed.

I second the idea of "stealing" Denis's two patches to block/aio_task 
and only adding the mutex (plus qatomic_read/set) here.

Paolo

> Then we should modify comment 
> insid AioTaskPool structure.
> 
> Anyway, I think it's simpler to just have one QEMU_MUTEX_GUARD() for the 
> whole function.

Re: [PATCH 5/6] co-shared-resource: protect with a mutex

[Paolo Bonzini]

On 10/05/21 13:40, Vladimir Sementsov-Ogievskiy wrote:
> 
>> co-shared-resource is currently not thread-safe, as also reported
>> in co-shared-resource.h. Add a QemuMutex because co_try_get_from_shres
>> can also be invoked from non-coroutine context.
> 
> But it doesn't. It's called only from co_get_from_shres(). So, better 
> make it a static function first.

It's a sensible interface though.  It lets you sleep or retry in your 
own way if you cannot get the resources, so (apart from the 
unlocked/locked confusion in the names) I like keeping it in the public API.

Paolo

Re: [PATCH 4/6] progressmeter: protect with a mutex

[Paolo Bonzini]

On 10/05/21 13:28, Vladimir Sementsov-Ogievskiy wrote:
> 
> 
> Could we instead add a c file and add the structure private? Then we'll 
> have progress_new() and progress_free() APIs instead.
> 
> This way, it would be a lot simpler to control that nobady use structure 
> fields directly.

I don't know...  I prefer embedding structs to heap allocation.

Paolo

Re: [PATCH 4/6] progressmeter: protect with a mutex

[Vladimir Sementsov-Ogievskiy]

11.05.2021 15:28, Paolo Bonzini wrote:
> On 10/05/21 13:28, Vladimir Sementsov-Ogievskiy wrote:
>>
>>
>> Could we instead add a c file and add the structure private? Then we'll have progress_new() and progress_free() APIs instead.
>>
>> This way, it would be a lot simpler to control that nobady use structure fields directly.
> 
> I don't know...  I prefer embedding structs to heap allocation.
> 


I agree that embedding looks better from allocation point of view.. But IMHO encapsulation guarantee of private structure is better feature. Especially when we have getter/setter methods. Even the patch itself is example: to review this carefully, I should somehow check that every use of the structure is updated (grepping the code, or maybe change field name and recompile, to catch every occurrence of it).. And if patch makes structure private and adds setters/getters, it can be reviewed without applying.

And we have to call _init/_destroy anyway, so it's not more complex to call _new/_free instead.

-- 
Best regards,
Vladimir

Re: [PATCH 0/6] block-copy: make helper APIs thread safe

[Stefan Hajnoczi]

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

On Mon, May 10, 2021 at 10:59:35AM +0200, Emanuele Giuseppe Esposito wrote:
> This serie of patches bring thread safety to the smaller APIs used by
> block-copy, namely ratelimit, progressmeter, co-shared-resource
> and aiotask.
> The end goal is to reduce the usage of the global
> AioContexlock in block-copy, by introducing smaller granularity
> locks thus on making the block layer thread safe. 

I'm not sure "global" is accurate. Block jobs can deal with arbitrary
AioContexts, not just a global AioContext. The lock is per-AioContext
rather than global. Or did I miss something?

[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 488 bytes --]

Re: [PATCH 6/6] aiopool: protect with a mutex

[Stefan Hajnoczi]

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

On Mon, May 10, 2021 at 10:59:41AM +0200, Emanuele Giuseppe Esposito wrote:
> Divide the fields in AioTaskPool in IN and Status, and
> introduce a CoQueue instead of .wait to take care of suspending
> and resuming the calling coroutine, and a lock to protect the
> busy_tasks counter accesses and the AioTask .ret field.

The thread-safety concerns with the aio_task.h API are unclear to me.
The API is designed to have a "main" coroutine that adds task
functions to the pool and waits for them to complete. Task functions
execute in coroutines (up to the pool's max_busy_tasks limit).

It seems like the API was designed to be called only from its main
coroutine so why make everything thread-safe? Is there a caller that
shares an AioTaskPool between threads? Or will the task functions switch
threads somehow?

What exactly is the new thread-safety model? Please document it.
Unfortunately aio_task.h doesn't have doc comments already but it will
be necessary if there are thread-safety concerns.

[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 488 bytes --]

Re: [PATCH 5/6] co-shared-resource: protect with a mutex

[Stefan Hajnoczi]

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

On Mon, May 10, 2021 at 10:59:40AM +0200, Emanuele Giuseppe Esposito wrote:
> co-shared-resource is currently not thread-safe, as also reported
> in co-shared-resource.h. Add a QemuMutex because co_try_get_from_shres
> can also be invoked from non-coroutine context.
> 
> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
> ---
>  util/qemu-co-shared-resource.c | 26 ++++++++++++++++++++++----
>  1 file changed, 22 insertions(+), 4 deletions(-)

Hmm...this thread-safety change is more fine-grained than I was
expecting. If we follow this strategy basically any data structure used
by coroutines needs its own fine-grained lock (like Java's Object base
class which has its own lock).

I'm not sure I like it since callers may still need coarser grained
locks to protect their own state or synchronize access to multiple
items of data. Also, some callers may not need thread-safety.

Can the caller to be responsible for locking instead (e.g. using
CoMutex)?

> diff --git a/util/qemu-co-shared-resource.c b/util/qemu-co-shared-resource.c
> index 1c83cd9d29..c455d02a1e 100644
> --- a/util/qemu-co-shared-resource.c
> +++ b/util/qemu-co-shared-resource.c
> @@ -32,6 +32,7 @@ struct SharedResource {
>      uint64_t available;
>  
>      CoQueue queue;
> +    QemuMutex lock;

Please add a comment indicating what this lock protects.

Thread safety should also be documented in the header file so API users
know what to expect.

[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 488 bytes --]

Re: [PATCH 4/6] progressmeter: protect with a mutex

[Stefan Hajnoczi]

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

On Mon, May 10, 2021 at 10:59:39AM +0200, Emanuele Giuseppe Esposito wrote:
> Progressmeter is protected by the AioContext mutex, which
> is taken by the block jobs and their caller (like blockdev).
> 
> We would like to remove the dependency of block layer code on the
> AioContext mutex, since most drivers and the core I/O code are already
> not relying on it.
> 
> The simplest thing to do is to add a mutex to be able to provide
> an accurate snapshot of the progress values to the caller.
> 
> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
> ---
>  blockjob.c                    | 33 +++++++++++++++++++++++++--------
>  include/qemu/progress_meter.h | 31 +++++++++++++++++++++++++++++++
>  job-qmp.c                     |  8 ++++++--
>  job.c                         |  3 +++
>  qemu-img.c                    |  9 ++++++---
>  5 files changed, 71 insertions(+), 13 deletions(-)

Unlike the next two patches where I had comments, here adding the lock
makes sense to me - the point of the progress meter is to report numbers
to the monitor thread so thread-safety is a key part of the API's value.

Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>

[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 488 bytes --]

Re: [PATCH 5/6] co-shared-resource: protect with a mutex

[Paolo Bonzini]

On 12/05/21 17:44, Stefan Hajnoczi wrote:
> If we follow this strategy basically any data structure used
> by coroutines needs its own fine-grained lock (like Java's Object base
> class which has its own lock).

Maybe not all, but only those that use CoQueue?  Interestingly, I was a 
bit less okay with ProgressMeter and didn't think twice about the other two.

Paolo

Re: [PATCH 5/6] co-shared-resource: protect with a mutex

[Emanuele Giuseppe Esposito]

On 12/05/2021 17:44, Stefan Hajnoczi wrote:
> On Mon, May 10, 2021 at 10:59:40AM +0200, Emanuele Giuseppe Esposito wrote:
>> co-shared-resource is currently not thread-safe, as also reported
>> in co-shared-resource.h. Add a QemuMutex because co_try_get_from_shres
>> can also be invoked from non-coroutine context.
>>
>> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
>> ---
>>   util/qemu-co-shared-resource.c | 26 ++++++++++++++++++++++----
>>   1 file changed, 22 insertions(+), 4 deletions(-)
> 
> Hmm...this thread-safety change is more fine-grained than I was
> expecting. If we follow this strategy basically any data structure used
> by coroutines needs its own fine-grained lock (like Java's Object base
> class which has its own lock).
> 
> I'm not sure I like it since callers may still need coarser grained
> locks to protect their own state or synchronize access to multiple
> items of data. Also, some callers may not need thread-safety.
> 
> Can the caller to be responsible for locking instead (e.g. using
> CoMutex)?

Right now co-shared-resource is being used only by block-copy, so I 
guess locking it from the caller or within the API won't really matter 
in this case.

One possible idea on how to delegate this to the caller without adding 
additional small lock/unlock in block-copy is to move co_get_from_shres 
in block_copy_task_end, and calling it only when a boolean passed to 
block_copy_task_end is true.

Otherwise make b_c_task_end always call co_get_from_shres and then 
include co_get_from_shres in block_copy_task_create, so that we always 
add and in case remove (if error) in the shared resource.

Something like:

diff --git a/block/block-copy.c b/block/block-copy.c
index 3a447a7c3d..1e4914b0cb 100644
--- a/block/block-copy.c
+++ b/block/block-copy.c
@@ -233,6 +233,7 @@ static coroutine_fn BlockCopyTask 
*block_copy_task_create(BlockCopyState *s,
      /* region is dirty, so no existent tasks possible in it */
      assert(!find_conflicting_task(s, offset, bytes));
      QLIST_INSERT_HEAD(&s->tasks, task, list);
+    co_get_from_shres(s->mem, task->bytes);
      qemu_co_mutex_unlock(&s->tasks_lock);

      return task;
@@ -269,6 +270,7 @@ static void coroutine_fn 
block_copy_task_end(BlockCopyTask *task, int ret)
          bdrv_set_dirty_bitmap(task->s->copy_bitmap, task->offset, 
task->bytes);
      }
      qemu_co_mutex_lock(&task->s->tasks_lock);
+    co_put_to_shres(task->s->mem, task->bytes);
      task->s->in_flight_bytes -= task->bytes;
      QLIST_REMOVE(task, list);
      progress_set_remaining(task->s->progress,
@@ -379,7 +381,6 @@ static coroutine_fn int 
block_copy_task_run(AioTaskPool *pool,

      aio_task_pool_wait_slot(pool);
      if (aio_task_pool_status(pool) < 0) {
-        co_put_to_shres(task->s->mem, task->bytes);
          block_copy_task_end(task, -ECANCELED);
          g_free(task);
          return -ECANCELED;
@@ -498,7 +499,6 @@ static coroutine_fn int 
block_copy_task_entry(AioTask *task)
      }
      qemu_mutex_unlock(&t->s->calls_lock);

-    co_put_to_shres(t->s->mem, t->bytes);
      block_copy_task_end(t, ret);

      return ret;
@@ -687,8 +687,6 @@ block_copy_dirty_clusters(BlockCopyCallState 
*call_state)

          trace_block_copy_process(s, task->offset);

-        co_get_from_shres(s->mem, task->bytes);
-
          offset = task_end(task);
          bytes = end - offset;




> 
>> diff --git a/util/qemu-co-shared-resource.c b/util/qemu-co-shared-resource.c
>> index 1c83cd9d29..c455d02a1e 100644
>> --- a/util/qemu-co-shared-resource.c
>> +++ b/util/qemu-co-shared-resource.c
>> @@ -32,6 +32,7 @@ struct SharedResource {
>>       uint64_t available;
>>   
>>       CoQueue queue;
>> +    QemuMutex lock;
> 
> Please add a comment indicating what this lock protects.
> 
> Thread safety should also be documented in the header file so API users
> know what to expect.

Will do, thanks.

Emanuele

Re: [PATCH 5/6] co-shared-resource: protect with a mutex

[Vladimir Sementsov-Ogievskiy via]

14.05.2021 17:10, Emanuele Giuseppe Esposito wrote:
> 
> 
> On 12/05/2021 17:44, Stefan Hajnoczi wrote:
>> On Mon, May 10, 2021 at 10:59:40AM +0200, Emanuele Giuseppe Esposito wrote:
>>> co-shared-resource is currently not thread-safe, as also reported
>>> in co-shared-resource.h. Add a QemuMutex because co_try_get_from_shres
>>> can also be invoked from non-coroutine context.
>>>
>>> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
>>> ---
>>>   util/qemu-co-shared-resource.c | 26 ++++++++++++++++++++++----
>>>   1 file changed, 22 insertions(+), 4 deletions(-)
>>
>> Hmm...this thread-safety change is more fine-grained than I was
>> expecting. If we follow this strategy basically any data structure used
>> by coroutines needs its own fine-grained lock (like Java's Object base
>> class which has its own lock).
>>
>> I'm not sure I like it since callers may still need coarser grained
>> locks to protect their own state or synchronize access to multiple
>> items of data. Also, some callers may not need thread-safety.
>>
>> Can the caller to be responsible for locking instead (e.g. using
>> CoMutex)?
> 
> Right now co-shared-resource is being used only by block-copy, so I guess locking it from the caller or within the API won't really matter in this case.
> 
> One possible idea on how to delegate this to the caller without adding additional small lock/unlock in block-copy is to move co_get_from_shres in block_copy_task_end, and calling it only when a boolean passed to block_copy_task_end is true.
> 
> Otherwise make b_c_task_end always call co_get_from_shres and then include co_get_from_shres in block_copy_task_create, so that we always add and in case remove (if error) in the shared resource.
> 
> Something like:
> 
> diff --git a/block/block-copy.c b/block/block-copy.c
> index 3a447a7c3d..1e4914b0cb 100644
> --- a/block/block-copy.c
> +++ b/block/block-copy.c
> @@ -233,6 +233,7 @@ static coroutine_fn BlockCopyTask *block_copy_task_create(BlockCopyState *s,
>       /* region is dirty, so no existent tasks possible in it */
>       assert(!find_conflicting_task(s, offset, bytes));
>       QLIST_INSERT_HEAD(&s->tasks, task, list);
> +    co_get_from_shres(s->mem, task->bytes);
>       qemu_co_mutex_unlock(&s->tasks_lock);
> 
>       return task;
> @@ -269,6 +270,7 @@ static void coroutine_fn block_copy_task_end(BlockCopyTask *task, int ret)
>           bdrv_set_dirty_bitmap(task->s->copy_bitmap, task->offset, task->bytes);
>       }
>       qemu_co_mutex_lock(&task->s->tasks_lock);
> +    co_put_to_shres(task->s->mem, task->bytes);
>       task->s->in_flight_bytes -= task->bytes;
>       QLIST_REMOVE(task, list);
>       progress_set_remaining(task->s->progress,
> @@ -379,7 +381,6 @@ static coroutine_fn int block_copy_task_run(AioTaskPool *pool,
> 
>       aio_task_pool_wait_slot(pool);
>       if (aio_task_pool_status(pool) < 0) {
> -        co_put_to_shres(task->s->mem, task->bytes);
>           block_copy_task_end(task, -ECANCELED);
>           g_free(task);
>           return -ECANCELED;
> @@ -498,7 +499,6 @@ static coroutine_fn int block_copy_task_entry(AioTask *task)
>       }
>       qemu_mutex_unlock(&t->s->calls_lock);
> 
> -    co_put_to_shres(t->s->mem, t->bytes);
>       block_copy_task_end(t, ret);
> 
>       return ret;
> @@ -687,8 +687,6 @@ block_copy_dirty_clusters(BlockCopyCallState *call_state)
> 
>           trace_block_copy_process(s, task->offset);
> 
> -        co_get_from_shres(s->mem, task->bytes);

we want to get from shres here, after possible call to block_copy_task_shrink(), as task->bytes may be reduced.

> -
>           offset = task_end(task);
>           bytes = end - offset;
> 
> 
> 
> 
>>
>>> diff --git a/util/qemu-co-shared-resource.c b/util/qemu-co-shared-resource.c
>>> index 1c83cd9d29..c455d02a1e 100644
>>> --- a/util/qemu-co-shared-resource.c
>>> +++ b/util/qemu-co-shared-resource.c
>>> @@ -32,6 +32,7 @@ struct SharedResource {
>>>       uint64_t available;
>>>       CoQueue queue;
>>> +    QemuMutex lock;
>>
>> Please add a comment indicating what this lock protects.
>>
>> Thread safety should also be documented in the header file so API users
>> know what to expect.
> 
> Will do, thanks.
> 
> Emanuele
> 


-- 
Best regards,
Vladimir

Re: [PATCH 5/6] co-shared-resource: protect with a mutex

[Emanuele Giuseppe Esposito]

On 14/05/2021 16:26, Vladimir Sementsov-Ogievskiy wrote:
> 14.05.2021 17:10, Emanuele Giuseppe Esposito wrote:
>>
>>
>> On 12/05/2021 17:44, Stefan Hajnoczi wrote:
>>> On Mon, May 10, 2021 at 10:59:40AM +0200, Emanuele Giuseppe Esposito 
>>> wrote:
>>>> co-shared-resource is currently not thread-safe, as also reported
>>>> in co-shared-resource.h. Add a QemuMutex because co_try_get_from_shres
>>>> can also be invoked from non-coroutine context.
>>>>
>>>> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
>>>> ---
>>>>   util/qemu-co-shared-resource.c | 26 ++++++++++++++++++++++----
>>>>   1 file changed, 22 insertions(+), 4 deletions(-)
>>>
>>> Hmm...this thread-safety change is more fine-grained than I was
>>> expecting. If we follow this strategy basically any data structure used
>>> by coroutines needs its own fine-grained lock (like Java's Object base
>>> class which has its own lock).
>>>
>>> I'm not sure I like it since callers may still need coarser grained
>>> locks to protect their own state or synchronize access to multiple
>>> items of data. Also, some callers may not need thread-safety.
>>>
>>> Can the caller to be responsible for locking instead (e.g. using
>>> CoMutex)?
>>
>> Right now co-shared-resource is being used only by block-copy, so I 
>> guess locking it from the caller or within the API won't really matter 
>> in this case.
>>
>> One possible idea on how to delegate this to the caller without adding 
>> additional small lock/unlock in block-copy is to move 
>> co_get_from_shres in block_copy_task_end, and calling it only when a 
>> boolean passed to block_copy_task_end is true.
>>
>> Otherwise make b_c_task_end always call co_get_from_shres and then 
>> include co_get_from_shres in block_copy_task_create, so that we always 
>> add and in case remove (if error) in the shared resource.
>>
>> Something like:
>>
>> diff --git a/block/block-copy.c b/block/block-copy.c
>> index 3a447a7c3d..1e4914b0cb 100644
>> --- a/block/block-copy.c
>> +++ b/block/block-copy.c
>> @@ -233,6 +233,7 @@ static coroutine_fn BlockCopyTask 
>> *block_copy_task_create(BlockCopyState *s,
>>       /* region is dirty, so no existent tasks possible in it */
>>       assert(!find_conflicting_task(s, offset, bytes));
>>       QLIST_INSERT_HEAD(&s->tasks, task, list);
>> +    co_get_from_shres(s->mem, task->bytes);
>>       qemu_co_mutex_unlock(&s->tasks_lock);
>>
>>       return task;
>> @@ -269,6 +270,7 @@ static void coroutine_fn 
>> block_copy_task_end(BlockCopyTask *task, int ret)
>>           bdrv_set_dirty_bitmap(task->s->copy_bitmap, task->offset, 
>> task->bytes);
>>       }
>>       qemu_co_mutex_lock(&task->s->tasks_lock);
>> +    co_put_to_shres(task->s->mem, task->bytes);
>>       task->s->in_flight_bytes -= task->bytes;
>>       QLIST_REMOVE(task, list);
>>       progress_set_remaining(task->s->progress,
>> @@ -379,7 +381,6 @@ static coroutine_fn int 
>> block_copy_task_run(AioTaskPool *pool,
>>
>>       aio_task_pool_wait_slot(pool);
>>       if (aio_task_pool_status(pool) < 0) {
>> -        co_put_to_shres(task->s->mem, task->bytes);
>>           block_copy_task_end(task, -ECANCELED);
>>           g_free(task);
>>           return -ECANCELED;
>> @@ -498,7 +499,6 @@ static coroutine_fn int 
>> block_copy_task_entry(AioTask *task)
>>       }
>>       qemu_mutex_unlock(&t->s->calls_lock);
>>
>> -    co_put_to_shres(t->s->mem, t->bytes);
>>       block_copy_task_end(t, ret);
>>
>>       return ret;
>> @@ -687,8 +687,6 @@ block_copy_dirty_clusters(BlockCopyCallState 
>> *call_state)
>>
>>           trace_block_copy_process(s, task->offset);
>>
>> -        co_get_from_shres(s->mem, task->bytes);
> 
> we want to get from shres here, after possible call to 
> block_copy_task_shrink(), as task->bytes may be reduced.

Ah right, I missed that. So I guess if we want the caller to protect 
co-shared-resource, get_from_shres stays where it is, and put_ instead 
can still go into task_end (with a boolean enabling it).

Thank you
Emanuele
> 
>> -
>>           offset = task_end(task);
>>           bytes = end - offset;
>>
>>
>>
>>
>>>
>>>> diff --git a/util/qemu-co-shared-resource.c 
>>>> b/util/qemu-co-shared-resource.c
>>>> index 1c83cd9d29..c455d02a1e 100644
>>>> --- a/util/qemu-co-shared-resource.c
>>>> +++ b/util/qemu-co-shared-resource.c
>>>> @@ -32,6 +32,7 @@ struct SharedResource {
>>>>       uint64_t available;
>>>>       CoQueue queue;
>>>> +    QemuMutex lock;
>>>
>>> Please add a comment indicating what this lock protects.
>>>
>>> Thread safety should also be documented in the header file so API users
>>> know what to expect.
>>
>> Will do, thanks.
>>
>> Emanuele
>>
> 
> 

Re: [PATCH 5/6] co-shared-resource: protect with a mutex

[Vladimir Sementsov-Ogievskiy via]

14.05.2021 17:32, Emanuele Giuseppe Esposito wrote:
> 
> 
> On 14/05/2021 16:26, Vladimir Sementsov-Ogievskiy wrote:
>> 14.05.2021 17:10, Emanuele Giuseppe Esposito wrote:
>>>
>>>
>>> On 12/05/2021 17:44, Stefan Hajnoczi wrote:
>>>> On Mon, May 10, 2021 at 10:59:40AM +0200, Emanuele Giuseppe Esposito wrote:
>>>>> co-shared-resource is currently not thread-safe, as also reported
>>>>> in co-shared-resource.h. Add a QemuMutex because co_try_get_from_shres
>>>>> can also be invoked from non-coroutine context.
>>>>>
>>>>> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
>>>>> ---
>>>>>   util/qemu-co-shared-resource.c | 26 ++++++++++++++++++++++----
>>>>>   1 file changed, 22 insertions(+), 4 deletions(-)
>>>>
>>>> Hmm...this thread-safety change is more fine-grained than I was
>>>> expecting. If we follow this strategy basically any data structure used
>>>> by coroutines needs its own fine-grained lock (like Java's Object base
>>>> class which has its own lock).
>>>>
>>>> I'm not sure I like it since callers may still need coarser grained
>>>> locks to protect their own state or synchronize access to multiple
>>>> items of data. Also, some callers may not need thread-safety.
>>>>
>>>> Can the caller to be responsible for locking instead (e.g. using
>>>> CoMutex)?
>>>
>>> Right now co-shared-resource is being used only by block-copy, so I guess locking it from the caller or within the API won't really matter in this case.
>>>
>>> One possible idea on how to delegate this to the caller without adding additional small lock/unlock in block-copy is to move co_get_from_shres in block_copy_task_end, and calling it only when a boolean passed to block_copy_task_end is true.
>>>
>>> Otherwise make b_c_task_end always call co_get_from_shres and then include co_get_from_shres in block_copy_task_create, so that we always add and in case remove (if error) in the shared resource.
>>>
>>> Something like:
>>>
>>> diff --git a/block/block-copy.c b/block/block-copy.c
>>> index 3a447a7c3d..1e4914b0cb 100644
>>> --- a/block/block-copy.c
>>> +++ b/block/block-copy.c
>>> @@ -233,6 +233,7 @@ static coroutine_fn BlockCopyTask *block_copy_task_create(BlockCopyState *s,
>>>       /* region is dirty, so no existent tasks possible in it */
>>>       assert(!find_conflicting_task(s, offset, bytes));
>>>       QLIST_INSERT_HEAD(&s->tasks, task, list);
>>> +    co_get_from_shres(s->mem, task->bytes);
>>>       qemu_co_mutex_unlock(&s->tasks_lock);
>>>
>>>       return task;
>>> @@ -269,6 +270,7 @@ static void coroutine_fn block_copy_task_end(BlockCopyTask *task, int ret)
>>>           bdrv_set_dirty_bitmap(task->s->copy_bitmap, task->offset, task->bytes);
>>>       }
>>>       qemu_co_mutex_lock(&task->s->tasks_lock);
>>> +    co_put_to_shres(task->s->mem, task->bytes);
>>>       task->s->in_flight_bytes -= task->bytes;
>>>       QLIST_REMOVE(task, list);
>>>       progress_set_remaining(task->s->progress,
>>> @@ -379,7 +381,6 @@ static coroutine_fn int block_copy_task_run(AioTaskPool *pool,
>>>
>>>       aio_task_pool_wait_slot(pool);
>>>       if (aio_task_pool_status(pool) < 0) {
>>> -        co_put_to_shres(task->s->mem, task->bytes);
>>>           block_copy_task_end(task, -ECANCELED);
>>>           g_free(task);
>>>           return -ECANCELED;
>>> @@ -498,7 +499,6 @@ static coroutine_fn int block_copy_task_entry(AioTask *task)
>>>       }
>>>       qemu_mutex_unlock(&t->s->calls_lock);
>>>
>>> -    co_put_to_shres(t->s->mem, t->bytes);
>>>       block_copy_task_end(t, ret);
>>>
>>>       return ret;
>>> @@ -687,8 +687,6 @@ block_copy_dirty_clusters(BlockCopyCallState *call_state)
>>>
>>>           trace_block_copy_process(s, task->offset);
>>>
>>> -        co_get_from_shres(s->mem, task->bytes);
>>
>> we want to get from shres here, after possible call to block_copy_task_shrink(), as task->bytes may be reduced.
> 
> Ah right, I missed that. So I guess if we want the caller to protect co-shared-resource, get_from_shres stays where it is, and put_ instead can still go into task_end (with a boolean enabling it).

honestly, I don't follow how it helps thread-safety

>>
>>> -
>>>           offset = task_end(task);
>>>           bytes = end - offset;
>>>
>>>
>>>
>>>
>>>>
>>>>> diff --git a/util/qemu-co-shared-resource.c b/util/qemu-co-shared-resource.c
>>>>> index 1c83cd9d29..c455d02a1e 100644
>>>>> --- a/util/qemu-co-shared-resource.c
>>>>> +++ b/util/qemu-co-shared-resource.c
>>>>> @@ -32,6 +32,7 @@ struct SharedResource {
>>>>>       uint64_t available;
>>>>>       CoQueue queue;
>>>>> +    QemuMutex lock;
>>>>
>>>> Please add a comment indicating what this lock protects.
>>>>
>>>> Thread safety should also be documented in the header file so API users
>>>> know what to expect.
>>>
>>> Will do, thanks.
>>>
>>> Emanuele
>>>
>>
>>
> 


-- 
Best regards,
Vladimir

Re: [PATCH 5/6] co-shared-resource: protect with a mutex

[Emanuele Giuseppe Esposito]

On 14/05/2021 17:30, Vladimir Sementsov-Ogievskiy wrote:
> 14.05.2021 17:32, Emanuele Giuseppe Esposito wrote:
>>
>>
>> On 14/05/2021 16:26, Vladimir Sementsov-Ogievskiy wrote:
>>> 14.05.2021 17:10, Emanuele Giuseppe Esposito wrote:
>>>>
>>>>
>>>> On 12/05/2021 17:44, Stefan Hajnoczi wrote:
>>>>> On Mon, May 10, 2021 at 10:59:40AM +0200, Emanuele Giuseppe 
>>>>> Esposito wrote:
>>>>>> co-shared-resource is currently not thread-safe, as also reported
>>>>>> in co-shared-resource.h. Add a QemuMutex because 
>>>>>> co_try_get_from_shres
>>>>>> can also be invoked from non-coroutine context.
>>>>>>
>>>>>> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
>>>>>> ---
>>>>>>   util/qemu-co-shared-resource.c | 26 ++++++++++++++++++++++----
>>>>>>   1 file changed, 22 insertions(+), 4 deletions(-)
>>>>>
>>>>> Hmm...this thread-safety change is more fine-grained than I was
>>>>> expecting. If we follow this strategy basically any data structure 
>>>>> used
>>>>> by coroutines needs its own fine-grained lock (like Java's Object base
>>>>> class which has its own lock).
>>>>>
>>>>> I'm not sure I like it since callers may still need coarser grained
>>>>> locks to protect their own state or synchronize access to multiple
>>>>> items of data. Also, some callers may not need thread-safety.
>>>>>
>>>>> Can the caller to be responsible for locking instead (e.g. using
>>>>> CoMutex)?
>>>>
>>>> Right now co-shared-resource is being used only by block-copy, so I 
>>>> guess locking it from the caller or within the API won't really 
>>>> matter in this case.
>>>>
>>>> One possible idea on how to delegate this to the caller without 
>>>> adding additional small lock/unlock in block-copy is to move 
>>>> co_get_from_shres in block_copy_task_end, and calling it only when a 
>>>> boolean passed to block_copy_task_end is true.
>>>>
>>>> Otherwise make b_c_task_end always call co_get_from_shres and then 
>>>> include co_get_from_shres in block_copy_task_create, so that we 
>>>> always add and in case remove (if error) in the shared resource.
>>>>
>>>> Something like:
>>>>
>>>> diff --git a/block/block-copy.c b/block/block-copy.c
>>>> index 3a447a7c3d..1e4914b0cb 100644
>>>> --- a/block/block-copy.c
>>>> +++ b/block/block-copy.c
>>>> @@ -233,6 +233,7 @@ static coroutine_fn BlockCopyTask 
>>>> *block_copy_task_create(BlockCopyState *s,
>>>>       /* region is dirty, so no existent tasks possible in it */
>>>>       assert(!find_conflicting_task(s, offset, bytes));
>>>>       QLIST_INSERT_HEAD(&s->tasks, task, list);
>>>> +    co_get_from_shres(s->mem, task->bytes);
>>>>       qemu_co_mutex_unlock(&s->tasks_lock);
>>>>
>>>>       return task;
>>>> @@ -269,6 +270,7 @@ static void coroutine_fn 
>>>> block_copy_task_end(BlockCopyTask *task, int ret)
>>>>           bdrv_set_dirty_bitmap(task->s->copy_bitmap, task->offset, 
>>>> task->bytes);
>>>>       }
>>>>       qemu_co_mutex_lock(&task->s->tasks_lock);
>>>> +    co_put_to_shres(task->s->mem, task->bytes);
>>>>       task->s->in_flight_bytes -= task->bytes;
>>>>       QLIST_REMOVE(task, list);
>>>>       progress_set_remaining(task->s->progress,
>>>> @@ -379,7 +381,6 @@ static coroutine_fn int 
>>>> block_copy_task_run(AioTaskPool *pool,
>>>>
>>>>       aio_task_pool_wait_slot(pool);
>>>>       if (aio_task_pool_status(pool) < 0) {
>>>> -        co_put_to_shres(task->s->mem, task->bytes);
>>>>           block_copy_task_end(task, -ECANCELED);
>>>>           g_free(task);
>>>>           return -ECANCELED;
>>>> @@ -498,7 +499,6 @@ static coroutine_fn int 
>>>> block_copy_task_entry(AioTask *task)
>>>>       }
>>>>       qemu_mutex_unlock(&t->s->calls_lock);
>>>>
>>>> -    co_put_to_shres(t->s->mem, t->bytes);
>>>>       block_copy_task_end(t, ret);
>>>>
>>>>       return ret;
>>>> @@ -687,8 +687,6 @@ block_copy_dirty_clusters(BlockCopyCallState 
>>>> *call_state)
>>>>
>>>>           trace_block_copy_process(s, task->offset);
>>>>
>>>> -        co_get_from_shres(s->mem, task->bytes);
>>>
>>> we want to get from shres here, after possible call to 
>>> block_copy_task_shrink(), as task->bytes may be reduced.
>>
>> Ah right, I missed that. So I guess if we want the caller to protect 
>> co-shared-resource, get_from_shres stays where it is, and put_ instead 
>> can still go into task_end (with a boolean enabling it).
> 
> honestly, I don't follow how it helps thread-safety

 From my understanding, the whole point here is to have no lock in 
co-shared-resource but let the caller take care of it (block-copy).

The above was just an idea on how to do it.

> 
>>>
>>>> -
>>>>           offset = task_end(task);
>>>>           bytes = end - offset;
>>>>
>>>>
>>>>
>>>>
>>>>>
>>>>>> diff --git a/util/qemu-co-shared-resource.c 
>>>>>> b/util/qemu-co-shared-resource.c
>>>>>> index 1c83cd9d29..c455d02a1e 100644
>>>>>> --- a/util/qemu-co-shared-resource.c
>>>>>> +++ b/util/qemu-co-shared-resource.c
>>>>>> @@ -32,6 +32,7 @@ struct SharedResource {
>>>>>>       uint64_t available;
>>>>>>       CoQueue queue;
>>>>>> +    QemuMutex lock;
>>>>>
>>>>> Please add a comment indicating what this lock protects.
>>>>>
>>>>> Thread safety should also be documented in the header file so API 
>>>>> users
>>>>> know what to expect.
>>>>
>>>> Will do, thanks.
>>>>
>>>> Emanuele
>>>>
>>>
>>>
>>
> 
> 

Re: [PATCH 5/6] co-shared-resource: protect with a mutex

[Paolo Bonzini]

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

Il ven 14 mag 2021, 16:10 Emanuele Giuseppe Esposito <eesposit@redhat.com>
ha scritto:

> > I'm not sure I like it since callers may still need coarser grained
> > locks to protect their own state or synchronize access to multiple
> > items of data. Also, some callers may not need thread-safety.
> >
> > Can the caller to be responsible for locking instead (e.g. using
> > CoMutex)?
>
> Right now co-shared-resource is being used only by block-copy, so I
> guess locking it from the caller or within the API won't really matter
> in this case.
>
> One possible idea on how to delegate this to the caller without adding
> additional small lock/unlock in block-copy is to move co_get_from_shres
> in block_copy_task_end, and calling it only when a boolean passed to
> block_copy_task_end is true.
>

The patch below won't work because qemu_co_queue_wait would have to unlock
the CoMutex; therefore you would have to pass it as an additional argument
to co_get_from_shres.

Overall, neither co_get_from_shres not AioTaskPool should be fast paths, so
using a local lock seems to produce the simplest API.

Paolo


> Otherwise make b_c_task_end always call co_get_from_shres and then
> include co_get_from_shres in block_copy_task_create, so that we always
> add and in case remove (if error) in the shared resource.
>
> Something like:
>
> diff --git a/block/block-copy.c b/block/block-copy.c
> index 3a447a7c3d..1e4914b0cb 100644
> --- a/block/block-copy.c
> +++ b/block/block-copy.c
> @@ -233,6 +233,7 @@ static coroutine_fn BlockCopyTask
> *block_copy_task_create(BlockCopyState *s,
>       /* region is dirty, so no existent tasks possible in it */
>       assert(!find_conflicting_task(s, offset, bytes));
>       QLIST_INSERT_HEAD(&s->tasks, task, list);
> +    co_get_from_shres(s->mem, task->bytes);
>       qemu_co_mutex_unlock(&s->tasks_lock);
>
>       return task;
> @@ -269,6 +270,7 @@ static void coroutine_fn
> block_copy_task_end(BlockCopyTask *task, int ret)
>           bdrv_set_dirty_bitmap(task->s->copy_bitmap, task->offset,
> task->bytes);
>       }
>       qemu_co_mutex_lock(&task->s->tasks_lock);
> +    co_put_to_shres(task->s->mem, task->bytes);
>       task->s->in_flight_bytes -= task->bytes;
>       QLIST_REMOVE(task, list);
>       progress_set_remaining(task->s->progress,
> @@ -379,7 +381,6 @@ static coroutine_fn int
> block_copy_task_run(AioTaskPool *pool,
>
>       aio_task_pool_wait_slot(pool);
>       if (aio_task_pool_status(pool) < 0) {
> -        co_put_to_shres(task->s->mem, task->bytes);
>           block_copy_task_end(task, -ECANCELED);
>           g_free(task);
>           return -ECANCELED;
> @@ -498,7 +499,6 @@ static coroutine_fn int
> block_copy_task_entry(AioTask *task)
>       }
>       qemu_mutex_unlock(&t->s->calls_lock);
>
> -    co_put_to_shres(t->s->mem, t->bytes);
>       block_copy_task_end(t, ret);
>
>       return ret;
> @@ -687,8 +687,6 @@ block_copy_dirty_clusters(BlockCopyCallState
> *call_state)
>
>           trace_block_copy_process(s, task->offset);
>
> -        co_get_from_shres(s->mem, task->bytes);
> -
>           offset = task_end(task);
>           bytes = end - offset;
>
>
>
>
> >
> >> diff --git a/util/qemu-co-shared-resource.c
> b/util/qemu-co-shared-resource.c
> >> index 1c83cd9d29..c455d02a1e 100644
> >> --- a/util/qemu-co-shared-resource.c
> >> +++ b/util/qemu-co-shared-resource.c
> >> @@ -32,6 +32,7 @@ struct SharedResource {
> >>       uint64_t available;
> >>
> >>       CoQueue queue;
> >> +    QemuMutex lock;
> >
> > Please add a comment indicating what this lock protects.
> >
> > Thread safety should also be documented in the header file so API users
> > know what to expect.
>
> Will do, thanks.
>
> Emanuele
>
>

[-- Attachment #2: Type: text/html, Size: 5230 bytes --]

Re: [PATCH 5/6] co-shared-resource: protect with a mutex

[Vladimir Sementsov-Ogievskiy via]

14.05.2021 20:28, Emanuele Giuseppe Esposito wrote:
> 
> 
> On 14/05/2021 17:30, Vladimir Sementsov-Ogievskiy wrote:
>> 14.05.2021 17:32, Emanuele Giuseppe Esposito wrote:
>>>
>>>
>>> On 14/05/2021 16:26, Vladimir Sementsov-Ogievskiy wrote:
>>>> 14.05.2021 17:10, Emanuele Giuseppe Esposito wrote:
>>>>>
>>>>>
>>>>> On 12/05/2021 17:44, Stefan Hajnoczi wrote:
>>>>>> On Mon, May 10, 2021 at 10:59:40AM +0200, Emanuele Giuseppe Esposito wrote:
>>>>>>> co-shared-resource is currently not thread-safe, as also reported
>>>>>>> in co-shared-resource.h. Add a QemuMutex because co_try_get_from_shres
>>>>>>> can also be invoked from non-coroutine context.
>>>>>>>
>>>>>>> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
>>>>>>> ---
>>>>>>>   util/qemu-co-shared-resource.c | 26 ++++++++++++++++++++++----
>>>>>>>   1 file changed, 22 insertions(+), 4 deletions(-)
>>>>>>
>>>>>> Hmm...this thread-safety change is more fine-grained than I was
>>>>>> expecting. If we follow this strategy basically any data structure used
>>>>>> by coroutines needs its own fine-grained lock (like Java's Object base
>>>>>> class which has its own lock).
>>>>>>
>>>>>> I'm not sure I like it since callers may still need coarser grained
>>>>>> locks to protect their own state or synchronize access to multiple
>>>>>> items of data. Also, some callers may not need thread-safety.
>>>>>>
>>>>>> Can the caller to be responsible for locking instead (e.g. using
>>>>>> CoMutex)?
>>>>>
>>>>> Right now co-shared-resource is being used only by block-copy, so I guess locking it from the caller or within the API won't really matter in this case.
>>>>>
>>>>> One possible idea on how to delegate this to the caller without adding additional small lock/unlock in block-copy is to move co_get_from_shres in block_copy_task_end, and calling it only when a boolean passed to block_copy_task_end is true.
>>>>>
>>>>> Otherwise make b_c_task_end always call co_get_from_shres and then include co_get_from_shres in block_copy_task_create, so that we always add and in case remove (if error) in the shared resource.
>>>>>
>>>>> Something like:
>>>>>
>>>>> diff --git a/block/block-copy.c b/block/block-copy.c
>>>>> index 3a447a7c3d..1e4914b0cb 100644
>>>>> --- a/block/block-copy.c
>>>>> +++ b/block/block-copy.c
>>>>> @@ -233,6 +233,7 @@ static coroutine_fn BlockCopyTask *block_copy_task_create(BlockCopyState *s,
>>>>>       /* region is dirty, so no existent tasks possible in it */
>>>>>       assert(!find_conflicting_task(s, offset, bytes));
>>>>>       QLIST_INSERT_HEAD(&s->tasks, task, list);
>>>>> +    co_get_from_shres(s->mem, task->bytes);
>>>>>       qemu_co_mutex_unlock(&s->tasks_lock);
>>>>>
>>>>>       return task;
>>>>> @@ -269,6 +270,7 @@ static void coroutine_fn block_copy_task_end(BlockCopyTask *task, int ret)
>>>>>           bdrv_set_dirty_bitmap(task->s->copy_bitmap, task->offset, task->bytes);
>>>>>       }
>>>>>       qemu_co_mutex_lock(&task->s->tasks_lock);
>>>>> +    co_put_to_shres(task->s->mem, task->bytes);
>>>>>       task->s->in_flight_bytes -= task->bytes;
>>>>>       QLIST_REMOVE(task, list);
>>>>>       progress_set_remaining(task->s->progress,
>>>>> @@ -379,7 +381,6 @@ static coroutine_fn int block_copy_task_run(AioTaskPool *pool,
>>>>>
>>>>>       aio_task_pool_wait_slot(pool);
>>>>>       if (aio_task_pool_status(pool) < 0) {
>>>>> -        co_put_to_shres(task->s->mem, task->bytes);
>>>>>           block_copy_task_end(task, -ECANCELED);
>>>>>           g_free(task);
>>>>>           return -ECANCELED;
>>>>> @@ -498,7 +499,6 @@ static coroutine_fn int block_copy_task_entry(AioTask *task)
>>>>>       }
>>>>>       qemu_mutex_unlock(&t->s->calls_lock);
>>>>>
>>>>> -    co_put_to_shres(t->s->mem, t->bytes);
>>>>>       block_copy_task_end(t, ret);
>>>>>
>>>>>       return ret;
>>>>> @@ -687,8 +687,6 @@ block_copy_dirty_clusters(BlockCopyCallState *call_state)
>>>>>
>>>>>           trace_block_copy_process(s, task->offset);
>>>>>
>>>>> -        co_get_from_shres(s->mem, task->bytes);
>>>>
>>>> we want to get from shres here, after possible call to block_copy_task_shrink(), as task->bytes may be reduced.
>>>
>>> Ah right, I missed that. So I guess if we want the caller to protect co-shared-resource, get_from_shres stays where it is, and put_ instead can still go into task_end (with a boolean enabling it).
>>
>> honestly, I don't follow how it helps thread-safety
> 
>  From my understanding, the whole point here is to have no lock in co-shared-resource but let the caller take care of it (block-copy).
> 
> The above was just an idea on how to do it.

But how moving co_put_to_shres() make it thread-safe? Nothing in block-copy is thread-safe yet..


-- 
Best regards,
Vladimir

Re: [PATCH 5/6] co-shared-resource: protect with a mutex

[Emanuele Giuseppe Esposito]

>>>>> we want to get from shres here, after possible call to 
>>>>> block_copy_task_shrink(), as task->bytes may be reduced.
>>>>
>>>> Ah right, I missed that. So I guess if we want the caller to protect 
>>>> co-shared-resource, get_from_shres stays where it is, and put_ 
>>>> instead can still go into task_end (with a boolean enabling it).
>>>
>>> honestly, I don't follow how it helps thread-safety
>>
>>  From my understanding, the whole point here is to have no lock in 
>> co-shared-resource but let the caller take care of it (block-copy).
>>
>> The above was just an idea on how to do it.
> 
> But how moving co_put_to_shres() make it thread-safe? Nothing in 
> block-copy is thread-safe yet..
> 
Sorry this is my bad, I did not explain it properly. If you look closely 
at the diff I sent, there are locks in a similar way of my block-copy 
initial patch. So I am essentially assuming that block-copy has already 
locks, and moving co_put_to_shres in block_copy_task_end has the purpose 
of moving shres "in a function that has a critical section".

>>>>>>> @@ -269,6 +270,7 @@ static void coroutine_fn 
>>>>>>> block_copy_task_end(BlockCopyTask *task, int ret)
>>>>>>>           bdrv_set_dirty_bitmap(task->s->copy_bitmap, 
>>>>>>> task->offset, task->bytes);
>>>>>>>       }
>>>>>>>       qemu_co_mutex_lock(&task->s->tasks_lock);
                ^^^ locks
>>>>>>> +    co_put_to_shres(task->s->mem, task->bytes);
>>>>>>>       task->s->in_flight_bytes -= task->bytes;
>>>>>>>       QLIST_REMOVE(task, list);
>>>>>>>       progress_set_remaining(task->s->progress,

		unlocks here (not shown in the diff)
	 }

Hopefully now it is clear. Apologies again for the confusion.

Emanuele

Re: [PATCH 5/6] co-shared-resource: protect with a mutex

[Vladimir Sementsov-Ogievskiy via]

15.05.2021 00:53, Emanuele Giuseppe Esposito wrote:
> 
>>>>>> we want to get from shres here, after possible call to block_copy_task_shrink(), as task->bytes may be reduced.
>>>>>
>>>>> Ah right, I missed that. So I guess if we want the caller to protect co-shared-resource, get_from_shres stays where it is, and put_ instead can still go into task_end (with a boolean enabling it).
>>>>
>>>> honestly, I don't follow how it helps thread-safety
>>>
>>>  From my understanding, the whole point here is to have no lock in co-shared-resource but let the caller take care of it (block-copy).
>>>
>>> The above was just an idea on how to do it.
>>
>> But how moving co_put_to_shres() make it thread-safe? Nothing in block-copy is thread-safe yet..
>>
> Sorry this is my bad, I did not explain it properly. If you look closely at the diff I sent, there are locks in a similar way of my block-copy initial patch. So I am essentially assuming that block-copy has already locks, and moving co_put_to_shres in block_copy_task_end has the purpose of moving shres "in a function that has a critical section".


Hm. Understand now. If you go this way, I'd better add a critical section and keep shres operations where they are now. We don't have shres operation in task creation, so should not have in task finalization. Shres operations are kept in seperate. It probably not bad to refactor it to make shres operations as a part of task manipulation functions, but it would require more accurate refactoring, simpler is to keep it as is.


-- 
Best regards,
Vladimir