lkml.org 
[lkml]   [2022]   [Nov]   [14]   [last100]   RSS Feed
Views: [wrap][no wrap]   [headers]  [forward] 
 
Messages in this thread
    /
    Date
    From
    SubjectRe: [PATCH] sbitmap: Use single per-bitmap counting to wake up queued tags
    Gabriel, when looking through this patch, I've noticed we can loose wakeups
    after your latest simplifications. See below for details:

    On Sat 05-11-22 19:10:55, Gabriel Krisman Bertazi wrote:
    > @@ -587,7 +571,7 @@ static struct sbq_wait_state *sbq_wake_ptr(struct sbitmap_queue *sbq)
    > for (i = 0; i < SBQ_WAIT_QUEUES; i++) {
    > struct sbq_wait_state *ws = &sbq->ws[wake_index];
    >
    > - if (waitqueue_active(&ws->wait) && atomic_read(&ws->wait_cnt)) {
    > + if (waitqueue_active(&ws->wait)) {
    > if (wake_index != atomic_read(&sbq->wake_index))
    > atomic_set(&sbq->wake_index, wake_index);
    > return ws;

    Neither sbq_wake_ptr() nor sbitmap_queue_wake_up() now increment the
    wake_index after performing the wakeup. Thus we would effectively keep
    waking tasks from a single waitqueue until it becomes empty and only then
    go for the next waitqueue. This creates unnecessary unfairness in task
    wakeups and even possible starvation issues. So I think we need to advance
    wake_index somewhere. Perhaps here before returning waitqueue.

    > @@ -599,83 +583,31 @@ static struct sbq_wait_state *sbq_wake_ptr(struct sbitmap_queue *sbq)
    > return NULL;
    > }
    >
    > -static bool __sbq_wake_up(struct sbitmap_queue *sbq, int *nr)
    > +void sbitmap_queue_wake_up(struct sbitmap_queue *sbq, int nr)
    > {
    > + unsigned int wake_batch = READ_ONCE(sbq->wake_batch);
    > + struct sbq_wait_state *ws = NULL;
    > + unsigned int wakeups;
    >
    > + if (!atomic_read(&sbq->ws_active))
    > + return;
    >
    > + atomic_add(nr, &sbq->completion_cnt);
    > + wakeups = atomic_read(&sbq->wakeup_cnt);
    >
    > do {
    > + if (atomic_read(&sbq->completion_cnt) - wakeups < wake_batch)
    > + return;
    >
    > + if (!ws) {
    > + ws = sbq_wake_ptr(sbq);
    > + if (!ws)
    > + return;
    > + }
    > + } while (!atomic_try_cmpxchg(&sbq->wakeup_cnt,
    > + &wakeups, wakeups + wake_batch));
    >
    > wake_up_nr(&ws->wait, wake_batch);

    Now this may be also problematic - when we were checking the number of woken
    waiters in the older version of the patch (for others: internal version of
    the patch) this was fine but now it may happen that the 'ws' we have
    selected has no waiters anymore. And in that case we need to find another
    waitqueue because otherwise we'd be loosing too many wakeups and we could
    deadlock. So I think this rather needs to be something like:

    do {
    if (atomic_read(&sbq->completion_cnt) - wakeups < wake_batch)
    return;
    } while (!atomic_try_cmpxchg(&sbq->wakeup_cnt,
    &wakeups, wakeups + wake_batch));

    do {
    ws = sbq_wake_ptr(sbq);
    if (!ws)
    return;
    } while (!wake_up_nr(&ws->wait, wake_batch));

    with our original version of wake_up_nr() returning number of woken
    waiters. What do you think?

    Honza
    --
    Jan Kara <jack@suse.com>
    SUSE Labs, CR

    \
     
     \ /
      Last update: 2022-11-14 14:23    [W:3.127 / U:0.096 seconds]
    ©2003-2020 Jasper Spaans|hosted at Digital Ocean and TransIP|Read the blog|Advertise on this site