From: "Darrick J. Wong" <djwong@kernel.org>
To: Dave Chinner <david@fromorbit.com>
Cc: linux-xfs@vger.kernel.org, hsiangkao@redhat.com
Subject: Re: [PATCH 1/7] workqueue: bound maximum queue depth
Date: Thu, 18 Mar 2021 18:45:58 -0700 [thread overview]
Message-ID: <20210319014558.GC1670408@magnolia> (raw)
In-Reply-To: <20210319013355.776008-2-david@fromorbit.com>
On Fri, Mar 19, 2021 at 12:33:49PM +1100, Dave Chinner wrote:
> From: Dave Chinner <dchinner@redhat.com>
>
> Existing users of workqueues have bound maximum queue depths in
> their external algorithms (e.g. prefetch counts). For parallelising
> work that doesn't have an external bound, allow workqueues to
> throttle incoming requests at a maximum bound. Bounded workqueues
> also need to distribute work over all worker threads themselves as
> there is no external bounding or worker function throttling
> provided.
>
> Existing callers are not throttled and retain direct control of
> worker threads, only users of the new create interface will be
> throttled and concurrency managed.
>
> Signed-off-by: Dave Chinner <dchinner@redhat.com>
> ---
> libfrog/workqueue.c | 42 +++++++++++++++++++++++++++++++++++++++---
> libfrog/workqueue.h | 4 ++++
> 2 files changed, 43 insertions(+), 3 deletions(-)
>
> diff --git a/libfrog/workqueue.c b/libfrog/workqueue.c
> index fe3de4289379..97f3bf76d9b9 100644
> --- a/libfrog/workqueue.c
> +++ b/libfrog/workqueue.c
> @@ -40,13 +40,21 @@ workqueue_thread(void *arg)
> }
>
> /*
> - * Dequeue work from the head of the list.
> + * Dequeue work from the head of the list. If the queue was
> + * full then send a wakeup if we're configured to do so.
> */
> assert(wq->item_count > 0);
> + if (wq->max_queued && wq->item_count == wq->max_queued)
> + pthread_cond_broadcast(&wq->queue_full);
FWIW I'll probably end up changing this to:
if (wq->max_queued)
pthread_cond_signal(&wq->queue_full);
To support multiple producers whenever I get around to posting the giant
scrub rewrite that is .... 272nd in line. In the meantime:
Reviewed-by: Darrick J. Wong <djwong@kernel.org>
--D
> wi = wq->next_item;
> wq->next_item = wi->next;
> wq->item_count--;
>
> + if (wq->max_queued && wq->next_item) {
> + /* more work, wake up another worker */
> + pthread_cond_signal(&wq->wakeup);
> + }
> pthread_mutex_unlock(&wq->lock);
>
> (wi->function)(wi->queue, wi->index, wi->arg);
> @@ -58,10 +66,11 @@ workqueue_thread(void *arg)
>
> /* Allocate a work queue and threads. Returns zero or negative error code. */
> int
> -workqueue_create(
> +workqueue_create_bound(
> struct workqueue *wq,
> void *wq_ctx,
> - unsigned int nr_workers)
> + unsigned int nr_workers,
> + unsigned int max_queue)
> {
> unsigned int i;
> int err = 0;
> @@ -70,12 +79,16 @@ workqueue_create(
> err = -pthread_cond_init(&wq->wakeup, NULL);
> if (err)
> return err;
> + err = -pthread_cond_init(&wq->queue_full, NULL);
> + if (err)
> + goto out_wake;
> err = -pthread_mutex_init(&wq->lock, NULL);
> if (err)
> goto out_cond;
>
> wq->wq_ctx = wq_ctx;
> wq->thread_count = nr_workers;
> + wq->max_queued = max_queue;
> wq->threads = malloc(nr_workers * sizeof(pthread_t));
> if (!wq->threads) {
> err = -errno;
> @@ -102,10 +115,21 @@ workqueue_create(
> out_mutex:
> pthread_mutex_destroy(&wq->lock);
> out_cond:
> + pthread_cond_destroy(&wq->queue_full);
> +out_wake:
> pthread_cond_destroy(&wq->wakeup);
> return err;
> }
>
> +int
> +workqueue_create(
> + struct workqueue *wq,
> + void *wq_ctx,
> + unsigned int nr_workers)
> +{
> + return workqueue_create_bound(wq, wq_ctx, nr_workers, 0);
> +}
> +
> /*
> * Create a work item consisting of a function and some arguments and schedule
> * the work item to be run via the thread pool. Returns zero or a negative
> @@ -140,6 +164,7 @@ workqueue_add(
>
> /* Now queue the new work structure to the work queue. */
> pthread_mutex_lock(&wq->lock);
> +restart:
> if (wq->next_item == NULL) {
> assert(wq->item_count == 0);
> ret = -pthread_cond_signal(&wq->wakeup);
> @@ -150,6 +175,16 @@ workqueue_add(
> }
> wq->next_item = wi;
> } else {
> + /* throttle on a full queue if configured */
> + if (wq->max_queued && wq->item_count == wq->max_queued) {
> + pthread_cond_wait(&wq->queue_full, &wq->lock);
> + /*
> + * Queue might be empty or even still full by the time
> + * we get the lock back, so restart the lookup so we do
> + * the right thing with the current state of the queue.
> + */
> + goto restart;
> + }
> wq->last_item->next = wi;
> }
> wq->last_item = wi;
> @@ -201,5 +236,6 @@ workqueue_destroy(
> free(wq->threads);
> pthread_mutex_destroy(&wq->lock);
> pthread_cond_destroy(&wq->wakeup);
> + pthread_cond_destroy(&wq->queue_full);
> memset(wq, 0, sizeof(*wq));
> }
> diff --git a/libfrog/workqueue.h b/libfrog/workqueue.h
> index a56d1cf14081..a9c108d0e66a 100644
> --- a/libfrog/workqueue.h
> +++ b/libfrog/workqueue.h
> @@ -31,10 +31,14 @@ struct workqueue {
> unsigned int thread_count;
> bool terminate;
> bool terminated;
> + int max_queued;
> + pthread_cond_t queue_full;
> };
>
> int workqueue_create(struct workqueue *wq, void *wq_ctx,
> unsigned int nr_workers);
> +int workqueue_create_bound(struct workqueue *wq, void *wq_ctx,
> + unsigned int nr_workers, unsigned int max_queue);
> int workqueue_add(struct workqueue *wq, workqueue_func_t fn,
> uint32_t index, void *arg);
> int workqueue_terminate(struct workqueue *wq);
> --
> 2.30.1
>
next prev parent reply other threads:[~2021-03-19 1:46 UTC|newest]
Thread overview: 25+ messages / expand[flat|nested] mbox.gz Atom feed top
2021-03-19 1:33 [PATCH 0/7] repair: Phase 6 performance improvements Dave Chinner
2021-03-19 1:33 ` [PATCH 1/7] workqueue: bound maximum queue depth Dave Chinner
2021-03-19 1:45 ` Darrick J. Wong [this message]
2021-03-19 1:33 ` [PATCH 2/7] repair: Protect bad inode list with mutex Dave Chinner
2021-03-19 18:20 ` Darrick J. Wong
2021-03-19 22:20 ` Dave Chinner
2021-03-19 1:33 ` [PATCH 3/7] repair: protect inode chunk tree records with a mutex Dave Chinner
2021-03-19 18:11 ` Darrick J. Wong
2021-03-19 1:33 ` [PATCH 4/7] repair: parallelise phase 6 Dave Chinner
2021-03-19 1:33 ` [PATCH 5/7] repair: don't duplicate names in " Dave Chinner
2021-03-19 1:33 ` [PATCH 6/7] repair: convert the dir byaddr hash to a radix tree Dave Chinner
2021-03-19 22:44 ` Darrick J. Wong
2021-03-19 1:33 ` [PATCH 7/7] repair: scale duplicate name checking in phase 6 Dave Chinner
2021-03-19 1:38 ` [PATCH 0/7] repair: Phase 6 performance improvements Gao Xiang
2021-03-19 18:22 ` Darrick J. Wong
2021-03-20 2:09 ` Gao Xiang
2021-03-24 1:26 ` Gao Xiang
2021-03-24 2:08 ` Darrick J. Wong
-- strict thread matches above, loose matches on Subject: below --
2020-10-22 5:15 Dave Chinner
2020-10-22 5:15 ` [PATCH 1/7] workqueue: bound maximum queue depth Dave Chinner
2020-10-22 5:54 ` Darrick J. Wong
2020-10-22 8:11 ` Dave Chinner
2020-10-25 4:41 ` Darrick J. Wong
2020-10-26 22:29 ` Dave Chinner
2020-10-26 22:40 ` Darrick J. Wong
2020-10-26 22:57 ` Dave Chinner
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
Avoid top-posting and favor interleaved quoting:
https://en.wikipedia.org/wiki/Posting_style#Interleaved_style
* Reply using the --to, --cc, and --in-reply-to
switches of git-send-email(1):
git send-email \
--in-reply-to=20210319014558.GC1670408@magnolia \
--to=djwong@kernel.org \
--cc=david@fromorbit.com \
--cc=hsiangkao@redhat.com \
--cc=linux-xfs@vger.kernel.org \
/path/to/YOUR_REPLY
https://kernel.org/pub/software/scm/git/docs/git-send-email.html
* If your mail client supports setting the In-Reply-To header
via mailto: links, try the mailto: link
Be sure your reply has a Subject: header at the top and a blank line
before the message body.
This is a public inbox, see mirroring instructions
for how to clone and mirror all data and code used for this inbox;
as well as URLs for NNTP newsgroup(s).